seqencoder.cc

来自「Motion JPEG编解码器源代码」· CC 代码 · 共 620 行 · 第 1/2 页

/* (C) 2000, 2001, 2005 Andrew Stevens 
 *  This file is free software; you can redistribute it
 *  and/or modify it under the terms of the GNU General Public License
 *  as published by the Free Software Foundation; either version 2 of
 *  the License, or (at your option) any later version.
 *
 *  This program 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 this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 * 02111-1307, USA.
 *
 */
/*
 * Disclaimer of Warranty
 *
 * These software programs are available to the user without any license fee or
 * royalty on an "as is" basis.  The MPEG Software Simulation Group disclaims
 * any and all warranties, whether express, implied, or statuary, including any
 * implied warranties or merchantability or of fitness for a particular
 * purpose.  In no event shall the copyright-holder be liable for any
 * incidental, punitive, or consequential damages of any kind whatsoever
 * arising from the use of these programs.
 *
 * This disclaimer of warranty extends to the user of these programs and user's
 * customers, employees, agents, transferees, successors, and assigns.
 *
 * The MPEG Software Simulation Group does not represent or warrant that the
 * programs furnished hereunder are free of infringement of any third-party
 * patents.
 *
 * Commercial implementations of MPEG-1 and MPEG-2 video, including shareware,
 * are subject to royalty fees to patent holders.  Many of these patents are
 * general enough such that they are unavoidable regardless of implementation
 * design.
 *
 */

#include "config.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <cassert>
#include "mjpeg_types.h"
#include "mjpeg_logging.h"
#include "mpeg2syntaxcodes.h"
#include "mpeg2encoder.hh"
#include "elemstrmwriter.hh"
#include "picturereader.hh"
#include "seqencoder.hh"
#include "ratectl.hh"
#include "tables.h"
#include "channel.hh"


// --------------------------------------------------------------------------------
//  Striped Encoding Job parallel despatch classes


struct EncoderJob
{
    EncoderJob() : shutdown( false ) {}
    void (MacroBlock::*encodingFunc)(); 
    Picture *picture;
    unsigned int stripe;
    bool shutdown;
};

class ShutdownJob : public EncoderJob
{
public:
    ShutdownJob()
        {
            shutdown = true;
        }
};



class Despatcher
{
public:
    Despatcher();
    ~Despatcher();
    void Init( unsigned int mb_width,
               unsigned int mb_height,
               unsigned int parallelism );
    void Despatch( Picture *picture, void (MacroBlock::*encodingFunc)() );
    void ParallelWorker();
    void WaitForCompletion();
private:
    static void *ParallelPerformWrapper(void *despatcher);

    unsigned int parallelism;
    unsigned int mb_width;
    unsigned int mb_height;
    vector<unsigned int> stripe_start;
    vector<unsigned int> stripe_length;
    Channel<EncoderJob> jobs;	
    pthread_t *worker_threads;
};

Despatcher::Despatcher() :
    worker_threads(0)
{}

void Despatcher::Init( unsigned int _mb_width,
                       unsigned int _mb_height,
                       unsigned int _parallelism )
{
    parallelism = _parallelism;
    mb_width = _mb_width;
    mb_height = _mb_height;

    if( !parallelism )
        return;

    unsigned int mb_in_stripe = 0;
    int i = 0;
    unsigned int pitch = mb_width / parallelism;
    for( int stripe = 0; stripe < parallelism; ++stripe )
    {
        stripe_start.push_back(mb_in_stripe);
        mb_in_stripe += pitch;
        stripe_length.push_back(pitch);
    }
    stripe_length.back() = mb_width - stripe_start.back();
	pthread_attr_t *pattr = NULL;

	/* For some Unixen we get a ridiculously small default stack size.
	   Hence we need to beef this up if we can.
	*/
#ifdef HAVE_PTHREADSTACKSIZE
#define MINSTACKSIZE 200000
	pthread_attr_t attr;
	size_t stacksize;

	pthread_attr_init(&attr);
	pthread_attr_getstacksize(&attr, &stacksize);

	if (stacksize < MINSTACKSIZE) {
		pthread_attr_setstacksize(&attr, MINSTACKSIZE);
	}

	pattr = &attr;
#endif
    worker_threads = new pthread_t[parallelism];
	for(i = 0; i < parallelism; ++i )
	{
        mjpeg_info("Creating worker thread" );
		if( pthread_create( &worker_threads[i], pattr, 
                            &Despatcher::ParallelPerformWrapper,
                            this ) != 0 )
		{
			mjpeg_error_exit1( "worker thread creation failed: %s", strerror(errno) );
		}
	}
}

Despatcher::~Despatcher()
{
    if( worker_threads != 0 )
    {
        WaitForCompletion();
        unsigned int i;
        ShutdownJob shutdownjob;

        for( i = 0; i < parallelism; ++i )
        {
            jobs.Put( shutdownjob );
        }
        for( i = 0; i < parallelism; ++i )
        {
            pthread_join( worker_threads[i], NULL );
        }
        delete [] worker_threads;
    }
}

void *Despatcher::ParallelPerformWrapper(void *despatcher)
{
    static_cast<Despatcher *>(despatcher)->ParallelWorker();
    return 0;
}

void Despatcher::ParallelWorker()
{
	EncoderJob job;
	mjpeg_debug( "Worker thread started" );
    pthread_setcanceltype( PTHREAD_CANCEL_ASYNCHRONOUS, NULL );

	for(;;)
	{
        // Get Job to do and do it!!
        jobs.Get( job );
        if( job.shutdown )
        {
            mjpeg_info ("SHUTDOWN worker" );
            pthread_exit( 0 );
        }
        vector<MacroBlock>::iterator stripe_begin;
        vector<MacroBlock>::iterator stripe_end;
        vector<MacroBlock>::iterator mbi;

        stripe_begin = job.picture->mbinfo.begin() + stripe_start[job.stripe];
        for( int row = 0; row < mb_height; ++row )
        {
            stripe_end = stripe_begin + stripe_length[job.stripe];
            for( mbi = stripe_begin; mbi < stripe_end; ++mbi )
            {
                (*mbi.*job.encodingFunc)();
            }
            stripe_begin += mb_width;
        }

    }
}

void Despatcher::Despatch(  Picture *picture,
                            void (MacroBlock::*encodingFunc)() )
{
    EncoderJob job;
    job.encodingFunc = encodingFunc;
    job.picture = picture;
    for( job.stripe = 0; job.stripe < parallelism; ++job.stripe )
    {
        jobs.Put( job );
    }
}

void Despatcher::WaitForCompletion()
{
    //
    // We know all despatched jobs have completed if the entire
    // pool of worker threads is waiting on the job despatch
    // channel
    //
    jobs.WaitUntilConsumersWaitingAtLeast( parallelism );
}






// --------------------------------------------------------------------------------
//  Sequence Encoder top-level class.
//



SeqEncoder::SeqEncoder( EncoderParams &_encparams,
                        PictureReader &_reader,
                        Quantizer &_quantizer,
                        ElemStrmWriter &_writer,
                        RateCtl    &_ratecontroller ) :
    encparams( _encparams ),
    reader( _reader ),
    quantizer( _quantizer ),
    writer( _writer ),
    ratecontroller( _ratecontroller ),
    despatcher( *new Despatcher ),
    ss( _encparams, _reader )

{
}

SeqEncoder::~SeqEncoder()
{
    delete &despatcher;
}



void SeqEncoder::EncodePicture(Picture *picture)
{
	mjpeg_debug("Start %d %c %d %d",
			   picture->decode, 
			   pict_type_char[picture->pict_type],
			   picture->temp_ref,
			   picture->present);


	if( picture->pict_struct != FRAME_PICTURE )
		mjpeg_debug("Field %s (%d)",
				   (picture->pict_struct == TOP_FIELD) ? "top" : "bot",
				   picture->pict_struct
			);

	picture->MotionSubSampledLum();

    if( encparams.encoding_parallelism > 0 )
    {
        despatcher.Despatch( picture, &MacroBlock::Encode );
        despatcher.WaitForCompletion();
    }
    else
    {
        picture->EncodeMacroBlocks();
    }

	picture->QuantiseAndEncode(ratecontroller);
	picture->Reconstruct();

	/* Handle second field of a frame that is being field encoded */