/*
    Copyright 2005-2010 Intel Corporation.  All Rights Reserved.

    This file is part of Threading Building Blocks.

    Threading Building Blocks is free software; you can redistribute it
    and/or modify it under the terms of the GNU General Public License
    version 2 as published by the Free Software Foundation.

    Threading Building Blocks is distributed in the hope that it will be
    useful, but WITHOUT ANY WARRANTY; without even the implied warranty
    of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with Threading Building Blocks; if not, write to the Free Software
    Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA

    As a special exception, you may use this file as part of a free software
    library without restriction.  Specifically, if other files instantiate
    templates or use macros or inline functions from this file, or you compile
    this file and link it with other files to produce an executable, this
    file does not by itself cause the resulting executable to be covered by
    the GNU General Public License.  This exception does not however
    invalidate any other reasons why the executable file might be covered by
    the GNU General Public License.
*/

#include "tbb/pipeline.h"
#include "tbb/spin_mutex.h"
#include "tbb/cache_aligned_allocator.h"
#include "itt_notify.h"


namespace tbb {

namespace internal {

//! This structure is used to store task information in a input buffer
struct task_info {
    void* my_object;
    //! Invalid unless a task went through an ordered stage.
    Token my_token;
    //! False until my_token is set.
    bool my_token_ready;
    //! True if my_object is valid.
    bool is_valid;
    //! Set to initial state (no object, no token)
    void reset() {
        my_object = NULL;
        my_token = 0;
        my_token_ready = false;
        is_valid = false;
    }
};
//! A buffer of input items for a filter.
/** Each item is a task_info, inserted into a position in the buffer corresponding to a Token. */
class input_buffer {
    friend class tbb::internal::pipeline_root_task;
    friend class tbb::thread_bound_filter;

    typedef  Token  size_type;

    //! Array of deferred tasks that cannot yet start executing. 
    task_info* array;

    //! Size of array
    /** Always 0 or a power of 2 */
    size_type array_size;

    //! Lowest token that can start executing.
    /** All prior Token have already been seen. */
    Token low_token;

    //! Serializes updates.
    spin_mutex array_mutex;

    //! Resize "array".
    /** Caller is responsible to acquiring a lock on "array_mutex". */
    void grow( size_type minimum_size );

    //! Initial size for "array"
    /** Must be a power of 2 */
    static const size_type initial_buffer_size = 4;

    //! Used only for out of order buffer.
    Token high_token;

    //! True for ordered filter, false otherwise. 
    bool is_ordered;

    //! True for thread-bound filter, false otherwise. 
    bool is_bound;
public:
    //! Construct empty buffer.
    input_buffer( bool is_ordered_, bool is_bound_ ) : 
            array(NULL), array_size(0),
            low_token(0), high_token(0), 
            is_ordered(is_ordered_), is_bound(is_bound_) {
        grow(initial_buffer_size);
        __TBB_ASSERT( array, NULL );
    }

    //! Destroy the buffer.
    ~input_buffer() {
        __TBB_ASSERT( array, NULL );
        cache_aligned_allocator<task_info>().deallocate(array,array_size);
        poison_pointer( array );
    }

    //! Put a token into the buffer.
    /** If task information was placed into buffer, returns true;
        otherwise returns false, informing the caller to create and spawn a task.
    */
    // Using template to avoid explicit dependency on stage_task
    template<typename StageTask>
    bool put_token( StageTask& putter ) {
        {
            spin_mutex::scoped_lock lock( array_mutex );
            Token token;
            if( is_ordered ) {
                if( !putter.my_token_ready ) {
                    putter.my_token = high_token++;
                    putter.my_token_ready = true;
                }
                token = putter.my_token;
            } else
                token = high_token++;
            __TBB_ASSERT( (tokendiff_t)(token-low_token)>=0, NULL );
            if( token!=low_token || is_bound ) {
                // Trying to put token that is beyond low_token.
                // Need to wait until low_token catches up before dispatching.
                if( token-low_token>=array_size ) 
                    grow( token-low_token+1 );
                ITT_NOTIFY( sync_releasing, this );
                putter.put_task_info(array[token&(array_size-1)]);
                return true;
            }
        }
        return false;
    }

    //! Note that processing of a token is finished.
    /** Fires up processing of the next token, if processing was deferred. */
    // Using template to avoid explicit dependency on stage_task
    template<typename StageTask>
    void note_done( Token token, StageTask& spawner ) {
        task_info wakee;
        wakee.reset();
        {
            spin_mutex::scoped_lock lock( array_mutex );
            if( !is_ordered || token==low_token ) {
                // Wake the next task
                task_info& item = array[++low_token & (array_size-1)];
                ITT_NOTIFY( sync_acquired, this );
                wakee = item;
                item.is_valid = false;
            }
        }
        if( wakee.is_valid )
            spawner.spawn_stage_task(wakee);
    }

#if __TBB_TASK_GROUP_CONTEXT
    //! The method destroys all data in filters to prevent memory leaks
    void clear( filter* my_filter ) {
        long t=low_token;
        for( size_type i=0; i<array_size; ++i, ++t ){
            task_info& temp = array[t&(array_size-1)];
            if (temp.is_valid ) {
                my_filter->finalize(temp.my_object);
                temp.is_valid = false;
            }
        }
    }
#endif

    bool return_item(task_info& info, bool advance) {
        spin_mutex::scoped_lock lock( array_mutex );
        task_info& item = array[low_token&(array_size-1)];
        ITT_NOTIFY( sync_acquired, this );
        if( item.is_valid ) {
            info = item;
            item.is_valid = false;
            if (advance) low_token++;
            return true;
        }
        return false;
    }

    void put_item( task_info& info ) {
        info.is_valid = true;
        spin_mutex::scoped_lock lock( array_mutex );
        Token token;
        if( is_ordered ) {
            if( !info.my_token_ready ) {
                info.my_token = high_token++;
                info.my_token_ready = true;
            }
            token = info.my_token;
        } else
            token = high_token++;
        __TBB_ASSERT( (tokendiff_t)(token-low_token)>=0, NULL );
        if( token-low_token>=array_size ) 
            grow( token-low_token+1 );
        ITT_NOTIFY( sync_releasing, this );
        array[token&(array_size-1)] = info;
    }
};

void input_buffer::grow( size_type minimum_size ) {
    size_type old_size = array_size;
    size_type new_size = old_size ? 2*old_size : initial_buffer_size;
    while( new_size<minimum_size ) 
        new_size*=2;
    task_info* new_array = cache_aligned_allocator<task_info>().allocate(new_size);
    task_info* old_array = array;
    for( size_type i=0; i<new_size; ++i )
        new_array[i].is_valid = false;
    long t=low_token;
    for( size_type i=0; i<old_size; ++i, ++t )
        new_array[t&(new_size-1)] = old_array[t&(old_size-1)];
    array = new_array;
    array_size = new_size;
    if( old_array )
        cache_aligned_allocator<task_info>().deallocate(old_array,old_size);
}

class stage_task: public task, public task_info {
private:
    friend class tbb::pipeline;
    pipeline& my_pipeline;
    filter* my_filter;  
    //! True if this task has not yet read the input.
    bool my_at_start;
public:
    //! Construct stage_task for first stage in a pipeline.
    /** Such a stage has not read any input yet. */
    stage_task( pipeline& pipeline ) :
        my_pipeline(pipeline), 
        my_filter(pipeline.filter_list),
        my_at_start(true)
    {
        task_info::reset();
    }
    //! Construct stage_task for a subsequent stage in a pipeline.
    stage_task( pipeline& pipeline, filter* filter_, const task_info& info ) :
        task_info(info),
        my_pipeline(pipeline), 
        my_filter(filter_),
        my_at_start(false)
    {}
    //! Roughly equivalent to the constructor of input stage task
    void reset() {
        task_info::reset();
        my_filter = my_pipeline.filter_list;
        my_at_start = true;
    }
    //! The virtual task execution method
    /*override*/ task* execute();
#if __TBB_TASK_GROUP_CONTEXT
    ~stage_task()    
    {
        if (my_filter && my_object && (my_filter->my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(4)) {
            __TBB_ASSERT(is_cancelled(), "Trying to finalize the task that wasn't cancelled");
            my_filter->finalize(my_object);
            my_object = NULL;
        }
    }
#endif // __TBB_TASK_GROUP_CONTEXT
    //! Creates and spawns stage_task from task_info
    void spawn_stage_task(const task_info& info)
    {
        stage_task* clone = new (allocate_additional_child_of(*parent())) 
                                stage_task( my_pipeline, my_filter, info );
        spawn(*clone);
    }
    //! Puts current task information
    void put_task_info(task_info &where_to_put ) {
        where_to_put.my_object = my_object;
        where_to_put.my_token = my_token;
        where_to_put.my_token_ready = my_token_ready;
        where_to_put.is_valid = true;
    }
};

task* stage_task::execute() {
    __TBB_ASSERT( !my_at_start || !my_object, NULL );
    __TBB_ASSERT( !my_filter->is_bound(), NULL );
    if( my_at_start ) {
        if( my_filter->is_serial() ) {
            my_object = (*my_filter)(my_object);
            if( my_object ) {
                if( my_filter->is_ordered() ) {
                    my_token = my_pipeline.token_counter++; // ideally, with relaxed semantics
                    my_token_ready = true;
                } else if( (my_filter->my_filter_mode & my_filter->version_mask) >= __TBB_PIPELINE_VERSION(5) ) {
                    if( my_pipeline.has_thread_bound_filters )
                        my_pipeline.token_counter++; // ideally, with relaxed semantics
                }
                if( !my_filter->next_filter_in_pipeline ) {
                    reset();
                    goto process_another_stage;
                } else {
                    ITT_NOTIFY( sync_releasing, &my_pipeline.input_tokens );
                    if( --my_pipeline.input_tokens>0 )
                        spawn( *new( allocate_additional_child_of(*parent()) ) stage_task( my_pipeline ) );
                }
            } else {
                my_pipeline.end_of_input = true; 
                return NULL;
            }
        } else /*not is_serial*/ {
            if( my_pipeline.end_of_input )
                return NULL;
            if( (my_filter->my_filter_mode & my_filter->version_mask) >= __TBB_PIPELINE_VERSION(5) ) {
                if( my_pipeline.has_thread_bound_filters )
                    my_pipeline.token_counter++;
            }
            ITT_NOTIFY( sync_releasing, &my_pipeline.input_tokens );
            if( --my_pipeline.input_tokens>0 )
                spawn( *new( allocate_additional_child_of(*parent()) ) stage_task( my_pipeline ) );
            my_object = (*my_filter)(my_object);
            if( !my_object ) {
                my_pipeline.end_of_input = true; 
                if( (my_filter->my_filter_mode & my_filter->version_mask) >= __TBB_PIPELINE_VERSION(5) ) {
                    if( my_pipeline.has_thread_bound_filters )
                        my_pipeline.token_counter--;
                }
                return NULL;
            }
        }
        my_at_start = false;
    } else {
        my_object = (*my_filter)(my_object);
        if( my_filter->is_serial() )
            my_filter->my_input_buffer->note_done(my_token, *this);
    }
    my_filter = my_filter->next_filter_in_pipeline; 
    if( my_filter ) {
        // There is another filter to execute.
        // Crank up priority a notch.
        add_to_depth(1);
        if( my_filter->is_serial() ) {
            // The next filter must execute tokens in order
            if( my_filter->my_input_buffer->put_token(*this) ){
                // Can't proceed with the same item
                if( my_filter->is_bound() ) {
                    // Find the next non-thread-bound filter
                    do {
                        my_filter = my_filter->next_filter_in_pipeline;
                    } while( my_filter && my_filter->is_bound() );
                    // Check if there is an item ready to process
                    if( my_filter && my_filter->my_input_buffer->return_item(*this, !my_filter->is_serial()) ) 
                        goto process_another_stage;
                } 
                my_filter = NULL; // To prevent deleting my_object twice if exception occurs
                return NULL;
            }
        }
    } else {
        // Reached end of the pipe.
        if( ++my_pipeline.input_tokens>1 || my_pipeline.end_of_input || my_pipeline.filter_list->is_bound() )
            return NULL; // No need to recycle for new input
        ITT_NOTIFY( sync_acquired, &my_pipeline.input_tokens );
        // Recycle as an input stage task.
        reset();
    }
process_another_stage:
    /* A semi-hackish way to reexecute the same task object immediately without spawning.
       recycle_as_continuation marks the task for future execution,
       and then 'this' pointer is returned to bypass spawning. */
    recycle_as_continuation();
    return this;
}

class pipeline_root_task: public task {
    pipeline& my_pipeline;
    bool do_segment_scanning;

    /*override*/ task* execute() {
        if( !my_pipeline.end_of_input )
            if( !my_pipeline.filter_list->is_bound() )
                if( my_pipeline.input_tokens > 0 ) {
                    recycle_as_continuation();
                    set_ref_count(1);
                    return new( allocate_child() ) stage_task( my_pipeline );
                }
        if( do_segment_scanning ) {
            filter* current_filter = my_pipeline.filter_list->next_segment;
            /* first non-thread-bound filter that follows thread-bound one 
            and may have valid items to process */
            filter* first_suitable_filter = current_filter;
            while( current_filter ) {
                __TBB_ASSERT( !current_filter->is_bound(), "filter is thread-bound?" );
                __TBB_ASSERT( current_filter->prev_filter_in_pipeline->is_bound(), "previous filter is not thread-bound?" );
                if( !my_pipeline.end_of_input
                    || (tokendiff_t)(my_pipeline.token_counter - current_filter->my_input_buffer->low_token) > 0 )
                {
                    task_info info;
                    info.reset();
                    if( current_filter->my_input_buffer->return_item(info, !current_filter->is_serial()) ) {
                        set_ref_count(1);
                        recycle_as_continuation();
                        return new( allocate_child() ) stage_task( my_pipeline, current_filter, info);
                    }
                    current_filter = current_filter->next_segment;
                    if( !current_filter ) {
                        if( !my_pipeline.end_of_input ) {
                            recycle_as_continuation();
                            return this;
                        }
                        current_filter = first_suitable_filter;
                        __TBB_Yield();
                    }
                } else { 
                    /* The preceding pipeline segment is empty. 
                    Fast-forward to the next post-TBF segment. */
                    first_suitable_filter = first_suitable_filter->next_segment;
                    current_filter = first_suitable_filter; 
                }
            } /* end of while */
            return NULL;
        } else { 
            if( !my_pipeline.end_of_input ) {
                recycle_as_continuation();
                return this;
            }
            return NULL;
        }
    }
public:
    pipeline_root_task( pipeline& pipeline ): my_pipeline(pipeline), do_segment_scanning(false)
    {
        __TBB_ASSERT( my_pipeline.filter_list, NULL );
        filter* first = my_pipeline.filter_list;
        if( (first->my_filter_mode & first->version_mask) >= __TBB_PIPELINE_VERSION(5) ) {
            // Scanning the pipeline for segments 
            filter* head_of_previous_segment = first;
            for(  filter* subfilter=first->next_filter_in_pipeline;
                  subfilter!=NULL;
                  subfilter=subfilter->next_filter_in_pipeline )
            {
                if( subfilter->prev_filter_in_pipeline->is_bound() && !subfilter->is_bound() ) {
                    do_segment_scanning = true;
                    head_of_previous_segment->next_segment = subfilter;
                    head_of_previous_segment = subfilter;
                }
            }
        }
    }
};

#if _MSC_VER && !defined(__INTEL_COMPILER)
    // Workaround for overzealous compiler warnings
    // Suppress compiler warning about constant conditional expression
    #pragma warning (disable: 4127)
#endif

// The class destroys end_counter and clears all input buffers if pipeline was cancelled.
class pipeline_cleaner: internal::no_copy {
    pipeline& my_pipeline;  
public:
    pipeline_cleaner(pipeline& _pipeline) : 
        my_pipeline(_pipeline)
    {}
    ~pipeline_cleaner(){
#if __TBB_TASK_GROUP_CONTEXT
        if (my_pipeline.end_counter->is_cancelled()) // Pipeline was cancelled
            my_pipeline.clear_filters(); 
#endif
        my_pipeline.end_counter = NULL;            
    }
};

} // namespace internal

void pipeline::inject_token( task& ) {
    __TBB_ASSERT(0,"illegal call to inject_token");
}

#if __TBB_TASK_GROUP_CONTEXT
void pipeline::clear_filters() {
    for( filter* f = filter_list; f; f = f->next_filter_in_pipeline ) {
        if ((f->my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(4))
            if( internal::input_buffer* b = f->my_input_buffer )
                b->clear(f);
    }
}
#endif

pipeline::pipeline() : 
    filter_list(NULL),
    filter_end(NULL),
    end_counter(NULL),
    end_of_input(false),
    has_thread_bound_filters(false)
{
    token_counter = 0;
    input_tokens = 0;
}

pipeline::~pipeline() {
    clear();
}

void pipeline::clear() {
    filter* next;
    for( filter* f = filter_list; f; f=next ) {
        if( internal::input_buffer* b = f->my_input_buffer ) {
            delete b; 
            f->my_input_buffer = NULL;
        }
        next=f->next_filter_in_pipeline;
        f->next_filter_in_pipeline = filter::not_in_pipeline();
        if ( (f->my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(3) ) {
            f->prev_filter_in_pipeline = filter::not_in_pipeline();
            f->my_pipeline = NULL;
        }
        if ( (f->my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(5) )
            f->next_segment = NULL;
    }
    filter_list = filter_end = NULL;
}

void pipeline::add_filter( filter& filter_ ) {
#if TBB_USE_ASSERT
    if ( (filter_.my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(3) ) 
        __TBB_ASSERT( filter_.prev_filter_in_pipeline==filter::not_in_pipeline(), "filter already part of pipeline?" );
    __TBB_ASSERT( filter_.next_filter_in_pipeline==filter::not_in_pipeline(), "filter already part of pipeline?" );
    __TBB_ASSERT( !end_counter, "invocation of add_filter on running pipeline" );
#endif    
    if ( (filter_.my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(3) ) {
        filter_.my_pipeline = this;
        filter_.prev_filter_in_pipeline = filter_end;
        if ( filter_list == NULL)
            filter_list = &filter_;
        else
            filter_end->next_filter_in_pipeline = &filter_;
        filter_.next_filter_in_pipeline = NULL;
        filter_end = &filter_;
    }
    else
    {
        if( !filter_end )
            filter_end = reinterpret_cast<filter*>(&filter_list);
        
        *reinterpret_cast<filter**>(filter_end) = &filter_;
        filter_end = reinterpret_cast<filter*>(&filter_.next_filter_in_pipeline);
        *reinterpret_cast<filter**>(filter_end) = NULL;
    }
    if( (filter_.my_filter_mode & filter_.version_mask) >= __TBB_PIPELINE_VERSION(5) ) {
        if( filter_.is_serial() ) {
            if( filter_.is_bound() )
                has_thread_bound_filters = true;
            filter_.my_input_buffer = new internal::input_buffer( filter_.is_ordered(), filter_.is_bound() );
        }
        else {
            if( filter_.prev_filter_in_pipeline && filter_.prev_filter_in_pipeline->is_bound() )
                filter_.my_input_buffer = new internal::input_buffer( false, false );
        }
    } else {
        if( filter_.is_serial() ) {
            filter_.my_input_buffer = new internal::input_buffer( filter_.is_ordered(), false );
        }
    }

}

void pipeline::remove_filter( filter& filter_ ) {
    if (&filter_ == filter_list) 
        filter_list = filter_.next_filter_in_pipeline;
    else {
        __TBB_ASSERT( filter_.prev_filter_in_pipeline, "filter list broken?" ); 
        filter_.prev_filter_in_pipeline->next_filter_in_pipeline = filter_.next_filter_in_pipeline;
    }
    if (&filter_ == filter_end)
        filter_end = filter_.prev_filter_in_pipeline;
    else {
        __TBB_ASSERT( filter_.next_filter_in_pipeline, "filter list broken?" ); 
        filter_.next_filter_in_pipeline->prev_filter_in_pipeline = filter_.prev_filter_in_pipeline;
    }
    if( internal::input_buffer* b = filter_.my_input_buffer ) {
        delete b; 
        filter_.my_input_buffer = NULL;
    }
    filter_.next_filter_in_pipeline = filter_.prev_filter_in_pipeline = filter::not_in_pipeline();
    if ( (filter_.my_filter_mode & filter::version_mask) >= __TBB_PIPELINE_VERSION(5) )
        filter_.next_segment = NULL;
    filter_.my_pipeline = NULL;
}

void pipeline::run( size_t max_number_of_live_tokens
#if __TBB_TASK_GROUP_CONTEXT
    , tbb::task_group_context& context
#endif
    ) {
    __TBB_ASSERT( max_number_of_live_tokens>0, "pipeline::run must have at least one token" );
    __TBB_ASSERT( !end_counter, "pipeline already running?" );
    if( filter_list ) {
        internal::pipeline_cleaner my_pipeline_cleaner(*this);
        end_of_input = false;
#if __TBB_TASK_GROUP_CONTEXT            
        end_counter = new( task::allocate_root(context) ) internal::pipeline_root_task( *this );
#else
        end_counter = new( task::allocate_root() ) internal::pipeline_root_task( *this );
#endif
        input_tokens = internal::Token(max_number_of_live_tokens);
        // Start execution of tasks
        task::spawn_root_and_wait( *end_counter );
    } 
}

#if __TBB_TASK_GROUP_CONTEXT
void pipeline::run( size_t max_number_of_live_tokens ) {
    if( filter_list ) {
        // Construct task group context with the exception propagation mode expected 
        // by the pipeline caller.
        uintptr_t ctx_traits = filter_list->my_filter_mode & filter::exact_exception_propagation ? 
                task_group_context::default_traits :
                task_group_context::default_traits & ~task_group_context::exact_exception;
        task_group_context context(task_group_context::bound, ctx_traits);
        run(max_number_of_live_tokens, context);
    }
}
#endif // __TBB_TASK_GROUP_CONTEXT

filter::~filter() {
    if ( (my_filter_mode & version_mask) >= __TBB_PIPELINE_VERSION(3) ) {
        if ( next_filter_in_pipeline != filter::not_in_pipeline() ) { 
            __TBB_ASSERT( prev_filter_in_pipeline != filter::not_in_pipeline(), "probably filter list is broken" );
            my_pipeline->remove_filter(*this);
        } else 
            __TBB_ASSERT( prev_filter_in_pipeline == filter::not_in_pipeline(), "probably filter list is broken" );
    } else {
        __TBB_ASSERT( next_filter_in_pipeline==filter::not_in_pipeline(), "cannot destroy filter that is part of pipeline" );
    }
}

thread_bound_filter::result_type thread_bound_filter::process_item() {
    return internal_process_item(true);
}

thread_bound_filter::result_type thread_bound_filter::try_process_item() {
    return internal_process_item(false);
}

thread_bound_filter::result_type thread_bound_filter::internal_process_item(bool is_blocking) {
    internal::task_info info;
    info.reset();
    
    if( !prev_filter_in_pipeline ) {
        if( my_pipeline->end_of_input )
            return end_of_stream;
        while( my_pipeline->input_tokens == 0 ) {
            if( is_blocking )
                __TBB_Yield();
            else
                return item_not_available;
        }
        info.my_object = (*this)(info.my_object);
        if( info.my_object ) {
            my_pipeline->input_tokens--;
            if( is_ordered() ) {
                info.my_token = my_pipeline->token_counter;
                info.my_token_ready = true;
            }
            my_pipeline->token_counter++; // ideally, with relaxed semantics
        } else {
            my_pipeline->end_of_input = true; 
            return end_of_stream; 
        }
    } else { /* this is not an input filter */
        while( !my_input_buffer->return_item(info, /*advance=*/true) ) {
            if( my_pipeline->end_of_input && my_input_buffer->low_token == my_pipeline->token_counter )
                return end_of_stream;
            if( is_blocking )
                __TBB_Yield();
            else
                return item_not_available;
        }
        info.my_object = (*this)(info.my_object);
    }
    if( next_filter_in_pipeline ) {
        next_filter_in_pipeline->my_input_buffer->put_item(info);
    } else {
        my_pipeline->input_tokens++;
    }

    return success;
}

} // tbb