include/ff/combine.hpp
/* 
 *
 * This function allows to combine two farms where the first one is an ordered farm.
 *
 *
 */
This function allows to combine two farms where the first one is an ordered farm.
This function allows to combine two farms where the first one is an ordered farm .
static inline const ff_pipeline combine_ofarm_farm(ff_farm& farm1, ff_farm& farm2) {
    ff_pipeline newpipe;
    if (!farm1.isOFarm()) {
        error("combine_ofarm_farm, the first farm is not an ordered farm");
        return newpipe;
    }
    if (farm2.isOFarm() && farm1.getNWorkers() == farm2.getNWorkers()) {   
        error("combine_ofarm_farm, two ordered farms with the same cardinality, the function cambine_farms_nf must be called explicitly\n");
        return newpipe;
    }
    ff_farm newfarm1;
    if (farm1.ondemand_buffer())
        newfarm1.set_scheduling_ondemand(farm1.ondemand_buffer());
    ordered_lb* _lb= new ordered_lb(farm1.getNWorkers());
    assert(_lb);
    const size_t memsize = farm1.getNWorkers() * (2*newfarm1.ondemand_buffer()+3)+ DEF_OFARM_ONDEMAND_MEMORY; 
    newfarm1.ordered_resize_memory(memsize);
    _lb->init(newfarm1.ordered_get_memory(), memsize);
    newfarm1.setlb(_lb, true);
    OrderedCollectorWrapper* cw = new OrderedCollectorWrapper(DEF_OFARM_ONDEMAND_MEMORY);
    assert(cw);
    ff_node* emitter1 = farm1.getEmitter();   
    if (emitter1) {
        newfarm1.add_emitter(emitter1);
        if (farm1.isset_cleanup_emitter()) {
            newfarm1.cleanup_emitter(true);
            farm1.cleanup_emitter(false);
        }
    }     
    const svector<ff_node*>& w1= farm1.getWorkers();
    std::vector<ff_node*> W1(w1.size());
    for(size_t i=0;i<w1.size();++i) {
        W1[i] = new OrderedWorkerWrapper(w1[i], farm1.isset_cleanup_workers());
        assert(W1[i]);
    }
    if (farm1.isset_cleanup_workers())
        farm1.cleanup_workers(false);
    newfarm1.add_workers(W1);
    newfarm1.cleanup_workers(true);
    ff_node* collector1 = farm1.getCollector();
    ff_node* emitter2 = farm2.getEmitter();    
    if (!collector1 && !emitter2) {
        farm2.change_emitter(cw, true);
    } else {
        if (!collector1) {
            ff_comb *comb = new ff_comb(cw, emitter2, 
                                        true, farm2.isset_cleanup_emitter());
            if (farm2.isset_cleanup_emitter()) 
                farm2.cleanup_emitter(false);
            farm2.change_emitter(comb, true);
        } else {
            if (!emitter2) {
                ff_comb *comb = new ff_comb(cw, collector1, 
                                            true, farm1.isset_cleanup_collector());
                if (farm1.isset_cleanup_collector()) 
                    farm1.cleanup_collector(false);
                farm2.change_emitter(comb, true);
            } else {
                ff_comb *comb0 = new ff_comb(collector1,emitter2,
                                             farm1.isset_cleanup_collector(),
                                             farm2.isset_cleanup_emitter());
                if (farm1.isset_cleanup_collector()) 
                    farm1.cleanup_collector(false);
                if (farm2.isset_cleanup_emitter()) 
                    farm2.cleanup_emitter(false);
                ff_comb *comb = new ff_comb(cw, comb0,
                                            true, true);
                farm2.change_emitter(comb, true);
            }
        }
    }
    newpipe.add_stage(newfarm1);
    newpipe.add_stage(farm2);
    return newpipe;
}
static inline const ff_pipeline combine_ofarm_farm ( ff_farm & farm1 , ff_farm & farm2 ) { ff_pipeline newpipe ; if ( ! farm1 . isOFarm ( ) ) { error ( " " ) ; return newpipe ; } if ( farm2 . isOFarm ( ) && farm1 . getNWorkers ( ) == farm2 . getNWorkers ( ) ) { error ( " \n " ) ; return newpipe ; } ff_farm newfarm1 ; if ( farm1 . ondemand_buffer ( ) ) newfarm1 . set_scheduling_ondemand ( farm1 . ondemand_buffer ( ) ) ; ordered_lb * _lb = new ordered_lb ( farm1 . getNWorkers ( ) ) ; assert ( _lb ) ; const size_t memsize = farm1 . getNWorkers ( ) * ( 2 * newfarm1 . ondemand_buffer ( ) + 3 ) + DEF_OFARM_ONDEMAND_MEMORY ; newfarm1 . ordered_resize_memory ( memsize ) ; _lb -> init ( newfarm1 . ordered_get_memory ( ) , memsize ) ; newfarm1 . setlb ( _lb , true ) ; OrderedCollectorWrapper * cw = new OrderedCollectorWrapper ( DEF_OFARM_ONDEMAND_MEMORY ) ; assert ( cw ) ; ff_node * emitter1 = farm1 . getEmitter ( ) ; if ( emitter1 ) { newfarm1 . add_emitter ( emitter1 ) ; if ( farm1 . isset_cleanup_emitter ( ) ) { newfarm1 . cleanup_emitter ( true ) ; farm1 . cleanup_emitter ( false ) ; } } const svector < ff_node * > & w1 = farm1 . getWorkers ( ) ; std :: vector < ff_node * > W1 ( w1 . size ( ) ) ; for ( size_t i = 0 ; i < w1 . size ( ) ; ++ i ) { W1 [ i ] = new OrderedWorkerWrapper ( w1 [ i ] , farm1 . isset_cleanup_workers ( ) ) ; assert ( W1 [ i ] ) ; } if ( farm1 . isset_cleanup_workers ( ) ) farm1 . cleanup_workers ( false ) ; newfarm1 . add_workers ( W1 ) ; newfarm1 . cleanup_workers ( true ) ; ff_node * collector1 = farm1 . getCollector ( ) ; ff_node * emitter2 = farm2 . getEmitter ( ) ; if ( ! collector1 && ! emitter2 ) { farm2 . change_emitter ( cw , true ) ; } else { if ( ! collector1 ) { ff_comb * comb = new ff_comb ( cw , emitter2 , true , farm2 . isset_cleanup_emitter ( ) ) ; if ( farm2 . isset_cleanup_emitter ( ) ) farm2 . cleanup_emitter ( false ) ; farm2 . change_emitter ( comb , true ) ; } else { if ( ! emitter2 ) { ff_comb * comb = new ff_comb ( cw , collector1 , true , farm1 . isset_cleanup_collector ( ) ) ; if ( farm1 . isset_cleanup_collector ( ) ) farm1 . cleanup_collector ( false ) ; farm2 . change_emitter ( comb , true ) ; } else { ff_comb * comb0 = new ff_comb ( collector1 , emitter2 , farm1 . isset_cleanup_collector ( ) , farm2 . isset_cleanup_emitter ( ) ) ; if ( farm1 . isset_cleanup_collector ( ) ) farm1 . cleanup_collector ( false ) ; if ( farm2 . isset_cleanup_emitter ( ) ) farm2 . cleanup_emitter ( false ) ; ff_comb * comb = new ff_comb ( cw , comb0 , true , true ) ; farm2 . change_emitter ( comb , true ) ; } } } newpipe . add_stage ( newfarm1 ) ; newpipe . add_stage ( farm2 ) ; return newpipe ; }
This function allows to combine two farms where the first one is an ordered farm.
This function allows to combine two farms where the first one is an ordered farm .
// here it would be possible to call directly the combine_farms_nf function but // since this kind of transformation may violates the ordering semantics, // the user must call it explicitly //newpipe.add_stage(combine_farms_nf(farm1,farm2)); // here we have that the first farm is an ordered farm and the second farm // is either a standard farm or is an ordered farm with a number of workers // that is different from the one of the first farm // emitter1  // workers1 // collector1 + emitter2