Subversion Repositories shark

/*

 * Copyright 1995, Brown University, Providence, RI

 *

 * Permission to use and modify this software and its documentation for

 * any purpose other than its incorporation into a commercial product is

 * hereby granted without fee.  Permission to copy and distribute this

 * software and its documentation only for non-commercial use is also

 * granted without fee, provided, however, that the above copyright notice

 * appear in all copies, that both that copyright notice and this permission

 * notice appear in supporting documentation, that the name of Brown

 * University not be used in advertising or publicity pertaining to

 * distribution of the software without specific, written prior permission,

 * and that the person doing the distribution notify Brown University of

 * such distributions outside of his or her organization. Brown University

 * makes no representations about the suitability of this software for

 * any purpose.  It is provided "as is" without express or implied warranty.

 * Brown University requests notification of any modifications to this

 * software or its documentation.

 *

 * Send the following redistribution information:

 *

 *      Name:

 *      Organization:

 *      Address (postal and/or electronic):

 *

 * To:

 *      Software Librarian

 *      Computer Science Department, Box 1910

 *      Brown University

 *      Providence, RI 02912

 *

 *              or

 *

 *      brusd@cs.brown.edu

 *

 * We will acknowledge all electronic notifications.

 */

static char Version[] =

   "$Id: threadmp.c,v 1.1.1.1 2002-03-29 14:12:59 pj Exp $";

static char Trick_to_use_Version() {char *unused = Version; return unused[0];}

/* -------------------------------------------------------------------------

 *

 * Original code by Tom Meyer

 * Altered by Loring Holden to work with MPEGmovie object

 *

 * ------------------------------------------------------------------------- */

#include "thread_mp.H"

#include <iostream.h>

#include <unistd.h>

#include <signal.h>

#include <stdlib.h>

#include <stdio.h>

#if defined(sol)

#elif defined(sgi)

usptr_t *THREADprocessor_arena;

#endif

/* -----------------------  Private Methods  ------------------------------- */

//

//  ------------  SOL-specific stuff ----------------------

//

#if defined(sol)

#ifdef CPLPLTHREADS

void THREAD_processor::startup()

{

   while (1) {

      sema->wait(__FILE__, __LINE__);

      _start_routine(_arg,_arg2);

      semaDone->post(__FILE__, __LINE__);

   }

}

int THREAD_processor::wait_for_done()

{

   semaDone->wait(__FILE__, __LINE__);

   running=0;

   return 1;

}

int THREAD_processor::still_running()

{

        //try_wait returns 0 if semaphore could be decremented

        //         returns 1 if semaphore could not be decremented

        if (semaDone->try_wait()) {

                return 1;

        } else {

                running=0;

                return 0;

        }

}

#else

void *THREADprocessor_startup_func(void *proc)

{

   ((THREADprocessor *)proc)->startup_func();

   return NULL;

}

void THREAD_processor::startup_func()

{

   while (1) {

      sema_wait(&sema);

      _start_routine(_arg,_arg2);

      sema_post(&semaDone);

   }

}

int THREAD_processor::wait_for_done()

{

   sema_wait(&semaDone);

   running=0;

   return 1;

}

int THREAD_processor::still_running()

{

        //try_wait returns 0 if semaphore could be decremented

        //         returns 1 if semaphore could not be decremented

        if (sema_trywait(&semaDone)) {

                return 1;

        } else {

                running=0;

                return 0;

        }

}

#endif

char new_name[255];

THREAD_processor::THREAD_processor(char *name, char *thrname)

#if CPLPLTHREADS

: Thread (thrname)

#endif

{

#if CPLPLTHREADS

   char *newname,*newname2;

   newname = (char *)malloc(sizeof(char)*(strlen(name) + strlen("/sema")));

   sprintf(newname, "%s/sema", name);

   newname2 = (char *)malloc(sizeof(char)*(strlen(name) + strlen("/doneSema")));

   sprintf(newname2, "%s/doneSema", name);

   sema = new Semaphore(0, newname);

   if (sema==NULL) {

      cerr << "Could not allocate semaphore (sema)" << endl<<flush;

   }

   semaDone = new Semaphore(0, newname2);

   if (semaDone==NULL) {

      cerr << "Could not allocate semaphore (semaDone)" << endl<<flush;

   }

   make_runnable();

#else

   sema_init(&sema, 0, USYNC_THREAD, NULL);

   sema_init(&semaDone, 0, USYNC_THREAD, NULL);

   thr_create(NULL, NULL, THREADprocessor_startup_func, (void *)this,

              THR_BOUND, &thr);

#endif

   running=0;

}

void THREAD_processor::start_execute(void *(*start_routine)(void *, void*),

                                    void *arg, void* arg2)

{

   if (running) {

     fprintf(stderr,"start_execute: already executing\n");

     fflush(stderr);

   }

//   cerr << "Waking up a process" << endl << flush;

   _start_routine = start_routine;

   _arg = arg;

   _arg2 = arg2;

   running=1;

#ifdef CPLPLTHREADS

   sema->post(__FILE__, __LINE__);

#else

   sema_post(&sema);

#endif

}

int THREAD_mp_manager::num_procs() const

{

   return sysconf(_SC_NPROCESSORS_ONLN);

}

THREAD_mp_manager::THREAD_mp_manager(int _procs)

{

}

#elif defined(sgi)

//

//  ------------  SGI-specific stuff ----------------------

//

void THREADprocessor_startup_func(void *proc)

{

   ((THREADprocessor *)proc)->startup();

}

void THREAD_processor::startup()

{

   while (1) {

      // Wait for us to get restarted

      uspsema(_sema);

      // Execute the routine that got left for us to do

      _start_routine(_arg,_arg2);

      // If someone's blocked on us, unblock them

      usvsema(semaDone);

   }

}

void THREAD_processor::start_execute(void *(*start_routine)(void *, void *),

                                    void *arg, void *arg2)

{

   _start_routine = start_routine;

   _arg = arg;

   _arg2 = arg2;

   usvsema(_sema);

   running=1;

}

int THREAD_processor::wait_for_done()

{

   // Block the calling process

   uspsema(semaDone);

   running=0;

   return 1;

}

int THREAD_processor::still_running()

{

        if (ustestsema(semaDone)){

                uspsema(semaDone);

                running=0;

                return 0;

        } else {

                return 1;

        }

}

THREAD_processor::THREAD_processor(char *, char *)

{

   _sema = usnewsema(THREADprocessor_arena,0);     // Create our semaphore

   semaDone = usnewsema(THREADprocessor_arena,0);

   if (_sema==NULL) {

        cerr << "problem allocating _sema" << endl;

   }

   if (semaDone==NULL) {

        cerr << "problem allocating semaDone" << endl;

   }

   // Share all vertual space of the parent, Synchronize file open table

   if (sproc(THREADprocessor_startup_func, PR_SADDR | PR_SFDS,

          (void *)this)==-1) {

        cerr << "sproc problem" << endl;

   }

}

int THREAD_mp_manager::num_procs() const

{

   return prctl(PR_MAXPPROCS);

}

THREAD_mp_manager::THREAD_mp_manager(int)

{

   // Set us up so that everyone dies simultaneously

   prctl(PR_SETEXITSIG, SIGQUIT);

   // Allocate shared memory arena 

   usconfig(CONF_INITUSERS, num_procs()*2 + 8);

   //Should not be a hard coded name in case we are running more than once

   //on the same machine

   char *tmpFile=tempnam("/tmp","THRED");

   THREADprocessor_arena = usinit(tmpFile);

   unlink(tmpFile);

   free(tmpFile);

   if (!THREADprocessor_arena) {

      cerr << "fatal error, shared memory arena could not be created"<<endl;

      exit(1);

   }

}

#else

//

//  ------------  Specific stuff for platforms without multi-threading support

//

THREAD_processor::THREAD_processor(char *, char *)

{

    running=0;

}

void THREAD_processor::start_execute(void *(*start_routine)(void *, void*),

     void *arg,void *arg2)

{

   start_routine(arg,arg2);

}

void THREAD_processor::startup()

{

}

int THREAD_processor::wait_for_done() {

   return 1;

}

int THREAD_processor::still_running()

{

   return 1;

}

THREAD_mp_manager::THREAD_mp_manager(int)

{

    cerr << "No multithreading..." << endl;

}

THREAD_mp_manager::num_procs() const

{

   return 1;

}

#endif

//

// General stuff for all platforms

//

void

THREADmp_manager::THREADmp_initproc(int i) {

   char *name, *thrname;

   name = (char *)malloc(sizeof(char)*50);

   thrname = (char *)malloc(sizeof(char)*50);

   sprintf(name, "Proc_%d", i);

   sprintf(thrname, "Proc_%d/thread", i);

   procs[i] = new THREADprocessor(name, thrname);

   enqThread(i);

}

void

THREADmp_manager::THREADmp_newproc()

{

   procs=(THREADprocessor **) realloc((void *) procs,++_num_procs*

       sizeof(THREADprocessor *));

   THREADmp_initproc(_num_procs-1);

}

THREADmp_manager::THREADmp_manager(int _procs)

: THREAD_mp_manager(_procs)

{

   cerr << "Creating mp manager\n";

   if (_procs == 0) {

      _num_procs = num_procs();

   } else {

      _num_procs = _procs;

   }

   cerr << "Num procs: " << num_procs() << endl << flush;

   procs = (THREADprocessor **)malloc(sizeof(THREADprocessor *) *_num_procs);

   for (int i = 0; i < _num_procs; i++) {

      THREADmp_initproc(i);

   }

   cerr << "Done\n";

}

void

THREADmp_manager::enqThread(int threadNumber)

{

        queue.enQueue(threadNumber);

}

int

THREADmp_manager::deqThread()

{

        int itemOk;

        int retVal=queue.deQueue(itemOk);

        if (!itemOk) {

                // This better allocate a thread, or we're not halting

                THREADmp_newproc();

                return deqThread();

        } else return retVal;

}

THREADmp_manager::~THREADmp_manager()

{

   for (int i = 0; i < _num_procs; i++)

      delete procs[i];

   free(procs);

}