スレッド・プログラミング

１つのプロセスの中に複数のスレッドを作って動作させることを マルチスレッドと言います。
これに対して、今までのプロセスは、１つのアドレス空間に１つのスレッドだけが動いているので、これを強調する時は、 シングルスレッドのプロセスと言います。

最近では、内部に並列性を含んでいたプログラムを自然に表現するためにも使われる。

■pthread

■pthread を利用したプログラムのコンパイル

pthread を利用したプログラムを書く時は、 pthread.h のヘッダ・ファイルをインクルードする必要があります。
またリンク時には、-lpthread を付ける必要があります。

% cc -o create create.c -lpthread 
% ./create 
...
%

■スレッドの生成・消滅

スレッドは、普通のプログラムのサブルーチン（Ｃ言語の関数、手続き) に近い。
サブルーチンの場合、呼び出すと、呼び出された方が動き、自分自身は止まる。
スレッドでは、新たにスレッドを生成した場合、生成した方と生成された方は、論理的には２つとも同時に動く。

■fork-join モデル

子スレッドは、pthread_exit() (トップの手続きからリターン)、親は、pthread_join() で子スレッドの終了を待ちます。

後で join する必要がない時には、pthread_detach() を使って切り離してもよい。
(join しなくてもゾンビにはならない)

/*
    create-join-2.c -- スレッドを作るプログラム
*/

#include <pthread.h>
#define MAX_LOOP    3
#define MAX_THD     5
void func1( int x );
void func2( int x );

main() {
    int       i;
    pthread_t t1[MAX_THD],t2[MAX_THD];

    for (i=0; i<MAX_THD; i++) {
       pthread_create( &t1[i], NULL, (void *)func1, (void *)i );
       pthread_create( &t2[i], NULL, (void *)func2, (void *)i );
    }
    printf("main()\n");
    for (i=0; i<MAX_THD; i++) {
       pthread_join( t1[i], NULL );
       pthread_join( t2[i], NULL );
    }
}

void func1( int x ) {
    int i ;

    for( i = 0 ; i<MAX_LOOP ; i++ ) {
       printf("func1( %d ): %d \n",x, i );
    }
}

void func2( int x ) {
    int i ;

    for( i = 0 ; i<MAX_LOOP ; i++ ) {
       printf("func2( %d ): %d \n",x, i );
    }
}

実行例。

% cc -o create_join_2 create_join_2.c -lpthread 
% ./create_join_2  
main()
func1( 0 ): 0
func1( 0 ): 1
func1( 0 ): 2
func2( 0 ): 0
func2( 0 ): 1
func2( 0 ): 2
func1( 1 ): 0
func1( 1 ): 1
func1( 1 ): 2
func2( 1 ): 0
func2( 1 ): 1
func2( 1 ): 2
func1( 2 ): 0
func1( 2 ): 1
func1( 2 ): 2
func2( 2 ): 0
func2( 2 ): 1
func2( 2 ): 2
func1( 3 ): 0
func1( 3 ): 1
func1( 3 ): 2
func2( 3 ): 0
func2( 3 ): 1
func2( 3 ): 2
func1( 4 ): 0
func1( 4 ): 1
func1( 4 ): 2
func2( 4 ): 0
func2( 4 ): 1
func2( 4 ): 2
%

マルチスレッド・プログラミングでは、 main 関数も１つのスレッドとして実行していると考える。これを 初期スレッド 、あるいは、 メインスレッド とよぶ。

pthread では、メインスレッド以外のスレッドは、 pthread_create() で作られる。

どういう順序で実行されるかは、決まっていない。
スレッドは、もともと順番を決めないような処理、 非同期的(asynchronous)な処理を表現するためのもの。

どうしても他のスレッドと同期を行なう必要がある場合は、 mutex や条件変数といった同期機能を使う。

pthread_create()で指定された関数から戻ると、そのスレッドは終了する。 pthread_exit() を呼び出してもよい。ただし、初期スレッドが終了すると、プロセス全体が終了する。 exit() システムコールを呼び出しても終了する。

■ スレッド・プログラ pthd

/********************************************************************
*
* compile with -D_POSIX_PTHREAD_SEMANTICS switch;
* required by sigwait()
*
* sigint thread handles delivery of signal. uses sigwait() to wait
* for SIGINT signal.
*
********************************************************************/
#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <synch.h>

static void    *threadTwo(void *);
static void    *threadThree(void *);
static void    *sigint(void *);

sigset_t       signalSet;

void * main(void) {
    pthread_t    t,t2,t3;

    thr_setconcurrency(3);
    sigfillset ( &signalSet );
    /*
     * Block signals in initial thread. New threads will
     * inherit this signal mask.
     */
    pthread_sigmask ( SIG_BLOCK, &signalSet, NULL );

    printf("Creating threads\n");

    /* POSIX thread create arguments:
     * thr_id, attr, strt_func, arg
     */
    pthread_create(&t, NULL, sigint, NULL);
    pthread_create(&t2, NULL, threadTwo, NULL);
    pthread_create(&t3, NULL, threadThree, NULL);

    printf("##################\n");
    printf("press CTRL-C to deliver SIGINT to sigint thread\n");
    printf("##################\n");

    pthread_exit((void *)0);
}
static void * threadTwo(void *arg) {
    printf("hello world, from threadTwo [tid: %d]\n",
                            pthread_self());
    printf("threadTwo [tid: %d} is now complete and exiting\n",
                            pthread_self());
    pthread_exit((void *)0);
}

static void * threadThree(void *arg) {
    printf("hello world, from threadThree [tid: %d]\n",
                            pthread_self());
    printf("threadThree [tid: %d} is now complete and exiting\n",
                            pthread_self());
    pthread_exit((void *)0);
}

void * sigint(void *arg) {
    int    sig;
    int    err;

    printf("thread sigint [tid: %d] awaiting SIGINT\n",
                            pthread_self());

    /* use POSIX sigwait() -- 2 args
     * signal set, signum
     */
    err = sigwait ( &signalSet, &sig );

    /* test for SIGINT; could catch other signals */
    if (err || sig != SIGINT)
        abort();

    printf("\nSIGINT signal %d caught by sigint thread [tid: %d]\n",
                            sig, pthread_self());
    pthread_exit((void *)0);
}

実行例。

% ./pthd  
Creating threads
thread sigint [tid: 4] awaiting SIGINT
hello world, from threadTwo [tid: 5]
threadTwo [tid: 5} is now complete and exiting
hello world, from threadThree [tid: 6]
threadThree [tid: 6} is now complete and exiting
##################
press CTRL-C to deliver SIGINT to sigint thread
##################



^C
SIGINT signal 2 caught by sigint thread [tid: 4]

■ スレッド・プログラム thdrun

以下にソースコードと Makefile を示します。動作は solaris 上で確認すみです。

 def.h

/*----------------------------
  FileName   : def.h
 *----------------------------*/
#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <pthread.h>
#include <signal.h>
#include <errno.h>
#include "lib.h"

void* server_thd(void *);
void* client_thd(void *);
void sigcatch(int);
int logger();
char *get_arg(int,char **);

#define SUCCESS     0
#define ERROR       1
#define MAX_ARGS    2
#define SZ_BUF    256
#define MAX_BUF   256
#define MAX_THDS   10
#define TM_SHORT    1
#define TM_MIDDLE   2
#define TM_LONG     5

#define VERSION        "v1.00"
#ifdef BROADCAST
#define DEF_LOG_FNAME  "thdrunb.log"
#else
#define DEF_LOG_FNAME  "thdrun.log"
#endif

/*------------------------------------------------------
     definition of mutex_lock
  ------------------------------------------------------*/
#define LOCK_NO       0      /* lock to global var thread_no       */
#define LOCK_SIG      1      /* lock to wait/signal         */
#define MAX_LOCK      2

#define SIGNAL_HANDLER  0
#define THD_SERVER      1
#define THD_CLIENT      2

 thdrun.c

/*------------------------------------------------------------------------
   FileName : thdrun.c

cc [ flag ... ] file ...  -lpthread

int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_wait(pthread_cond_t *cond,pthread_mutex_t *mutex);
int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex,
                           const struct timespec *abstime);
  ------------------------------------------------------------------------*/
#include "def.h"
#define SIGHANDLE
#define SIGSEND     SIGINT

typedef struct {
   short    threadno;       /* thread number   0 ~ 9    */
   short    id;             /* THD_SERVER/THD_CLIENT    */
   int      pthreadno;      /* real thread number (OS)  */
   char     date[SZ_DATE];  /* save area of date        */
}  thread_info;             /* control info of client/server thread  */

int            thread_no=0;
char           *pname,*logfile;
char           *arglist = "[-f LOG_FILE_NAME]";
FILE           *fp=0;
thread_info    thd_cli[MAX_THDS],thd_srv,signal_hnd;
pthread_cond_t cond;

/*-----------------------------
   Mutex Lock/Unlock
  -----------------------------*/
pthread_mutex_t       mutex[MAX_LOCK];
pthread_mutexattr_t   mutexattr[MAX_LOCK];
#define SIG_LOCK      &mutex[LOCK_SIG]

int mutexInit() {
   int i,err;

   for (i=0; i<MAX_LOCK; i++) {
     err   = pthread_mutexattr_init(&mutexattr[i]);
     if (err) die_err("pthread_mutexattr_init",ERR_MUTEX_AINI,"");
     err   = pthread_mutex_init(&mutex[i],&mutexattr[i]);
     if (err) die_err("pthread_mutex_init",ERR_MUTEX_INI,"");
   }
}

int mutexLock(int no) {
   int err;

   err   = pthread_mutex_lock(&mutex[no]);
   if (err) die_err("pthread_mutex_lock",ERR_MUTEX_LCK,"");
}

int mutexUnlock(int no) {
   int err;

   err   = pthread_mutex_unlock(&mutex[no]);
   if (err) die_err("pthread_mutex_unlock",ERR_MUTEX_ULK,"");
}

/*-------------------------
   Main
  ------------------------*/
int main(int argc,char * argv[]) {
  int            i,err;
  char           *log;
  pthread_attr_t threadAttr;      /* thread attribute */
  pthread_t      cli_thread[MAX_THDS],srvthread;

/* initial settings */
  pname   = argv[0];
  log     = get_arg(argc,argv);
  if (log != NULL)   logfile = log;
  else               logfile = DEF_LOG_FNAME;
  fp = fopen(logfile,"a+");
  if (! fp)   die_err("fopen",ERR_OPEN,logfile);

#ifdef SIGHANDLE
  signal(SIGSEND,sigcatch);
#endif

/* prepare to create threads */
  mutexInit();

/* create threads              */
  for (i=0; i<MAX_THDS; i++) {
    thd_cli[i].threadno = i;
    thd_cli[i].id       = THD_CLIENT;
    err = pthread_create(&cli_thread[i], NULL , client_thd,
                         (void *) &thd_cli[i]);
    if (err)  die_err("pthread_create",ERR_PTHD_CRE,"client_thd");
  }
  thd_srv.id = THD_SERVER;
  err        = pthread_create(&srvthread, NULL , server_thd, (void *) &thd_srv);
  if (err)  die_err("pthread_create",ERR_PTHD_CRE,"server_thd");

  sleep(10);

  for (i=0; i<MAX_THDS; i++) {
    err = pthread_join(cli_thread[i], (void **) NULL);
    if (err)   die_err("pthread_join",ERR_PTHD_JOIN,"");
  }
  err = pthread_join(srvthread, (void **) NULL);
  if (err)   die_err("pthread_join",ERR_PTHD_JOIN,"");
}

/*-------------------------------------------------------------------------
   Function : check thread_no. if changes ,then send SIGNAL to client_thd
  -------------------------------------------------------------------------*/
void* server_thd(void *arg) {
  int           ret,prev_thdno=-1;
  unsigned int  seconds;
  pthread_t     myID;
  thread_info   *thd_info;

  /* initialize       */
  thd_info = (thread_info *) arg;
  myID     = pthread_self();
  printf("Server Thread %d created\n", myID),fflush(stdout);
  seconds = TM_SHORT;

  for (;;) {
    printf("Server lock\n"),fflush(stdout);
    mutexLock(LOCK_NO);
    if (thread_no != prev_thdno)  {
#ifdef BROADCAST
      ret = pthread_cond_broadcast(&cond);
#else
      ret = pthread_cond_signal(&cond);
#endif
      if (ret) die_err("pthread_cond_signal",ERR_SIGSEND,"Server");
      logger(thd_info,NULL);
      prev_thdno = thread_no;
    }
    mutexUnlock(LOCK_NO);
    sleep(seconds);
  }
}

/*-------------------------------------------------------------------------
   Function : Client thread. if exist data,then send SIGNAL
  -------------------------------------------------------------------------*/
void* client_thd(void *arg) {
  int       cur,thdno,err,ret,signo;
  char      *cp;
  pthread_t   myID;
  sigset_t    sigset;
  thread_info *thd_info;

/* initialize       */
  myID = pthread_self();
  printf("Client Thread %d  created\n", myID);
  fflush(stdout);
  thd_info            = (thread_info *) arg;
  thd_info->pthreadno = myID;
  thdno               = thd_info->threadno;

/* wait until to receive signal   */
  for (;;) {
    ret = pthread_cond_wait(&cond,&mutex[LOCK_SIG]);
    if (ret)    die_err("pthread_cond_wait",ERR_SIGWAIT,"client_thd");
/* catch signal      */
    printf("Client thread_%d(%d)  received signal\n",thdno,myID),fflush(stdout);

/* get thread_no record    */
     mutexLock(LOCK_NO);
     cur = thread_no % MAX_THDS;
     if (cur == thdno)   {
        logger(thd_info,thread_no);
        thread_no ++;
     }
     else  {
        printf("Client : I'm not thread_%d(%d). I'm thread_%d.\n",
                cur,thread_no,thdno);
     }
     mutexUnlock(LOCK_NO);
  }
}

char *msg_c = "%s  Client_thd: %d(%2d)  Add thread_no  (%d -> %d)\n";
char *logmsg[]={"%s  Signal_handler:    Received SIGINT(CNTL.C).\n",
#ifdef BROADCAST
                "%s  Server_thd:        Send the signal to all clients.\n"};
#else
                "%s  Server_thd:        Send the signal to client.\n"};
#endif

int logger(thread_info *info,int cur_thread_no) {
  char buf[SZ_BUF];
  int  stt;

  if (!fp)  die_err("logger",ERR_OPEN,"");

  switch(info->id) {
  case THD_CLIENT:
    sprintf(buf,msg_c,mkdate(info->date,SZ_DATE),
                info->threadno,info->pthreadno,
                cur_thread_no,cur_thread_no+1);
    break;

  case THD_SERVER:
  case SIGNAL_HANDLER:
    sprintf(buf,logmsg[info->id],mkdate(info->date,SZ_DATE));
    break;
  }

  stt = fprintf(fp,buf);
  if (stt < 0)   die_err("logger",ERR_FPRINTF,"");
  fflush(fp);
}

#ifdef SIGHANDLE
void sigcatch(int sig) {
  signal(SIGSEND,SIG_IGN);
  printf("Signal %d received\n",sig);
  signal_hnd.id = SIGNAL_HANDLER;
  logger(&signal_hnd,0);
  exit(SUCCESS);
}
#endif

extern char *optarg;
extern int  optind;
char *u_opt = "vhf:";

/*----------------------------------------------------------
   Function : analize input argumets
   Param    : argc(I) = N(arguments)
              argv(I) = argumets array
   Return   : table_no to make
  ----------------------------------------------------------*/
char *get_arg(int argc,char *argv[]) {
  int  c;
  char *rtn = NULL;
  static char *func = "get_arg";

  while((c=getopt(argc,argv,u_opt)) != EOF)
  switch(c) {
      case 'f':
        rtn = optarg;
        break;
      case 'v':
        printf("\n%s %s\n",pname,VERSION);
        exit(SUCCESS);
        break;
      case 'h':
        usage(pname,arglist);
        exit(SUCCESS);
        break;
      default:
        usage(pname,arglist);
        exit(ERROR);
        break;
  }
  return rtn;
}

 lib.h 

/*-------------------------------
  FileName   : lib.h
  -------------------------------*/
#define SZ_DATE     28      /* mkdate:length of date (yy/mm/dd hh:mm:ss)   */
#define SZ_MSGBUF  126      /* error: message buf size              */

char *mkdate();
void usage();

/*------------------------------------------------------
   Usage of die_err and warn_err
   Calling : die_err("LIB_NAME",ERR_CODE,"FUNCTION_NAME")
   Action  : print error messages and exit with ERR_CODE
             SystemError  : strerror
             GeneralError : put err_msg[]

   Calling : warn_err("LIB_NAME",ERR_CODE,"FUNCTION_NAME")
   Action  : print error messages and return
  ------------------------------------------------------*/
void error(char *,int,char *,short);
#define DIE_ERR   1     /* fatal error  */
#define WARN_ERR  0     /* warning error  */
#define die_err(msg,no,info)   error(msg,no,info,DIE_ERR);
#define warn_err(msg,no,info)  error(msg,no,info,WARN_ERR);

/*
   definition of Error no
*/
/* System error : these are from 1 to ESTALE(151)   /usr/include/sys/errno.h */
#define ERR_MUTEX_AINI   1
#define ERR_MUTEX_INI    2
#define ERR_MUTEX_LCK    3
#define ERR_MUTEX_ULK    4

#define ERR_SIGWAIT     11
#define ERR_SIGSEND     12

#define ERR_MALLOC      20
#define ERR_LSEEK       21
#define ERR_READ        22
#define ERR_WRITE       23
#define ERR_OPEN        24

#define ERR_PTHD_INIT   30
#define ERR_PTHD_ATTR   31
#define ERR_PTHD_CRE    32
#define ERR_PTHD_JOIN   33

#define ERR_FPRINTF     40

/* general error : these are ESTALE to 255  /usr/include/sys/errno.h    */
#define ERR_INIT      200
#define ERR_END       255
#define ERR_UTL_S      ERR_INIT
#define ERR_UTL_E      ERR_END


 lib.c 

#include <stdio.h>
#include <time.h>
#include <unistd.h>
#include <signal.h>
#include <errno.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/types.h>
#include "lib.h"

/*---------------------------------------------------------
   Function : make current time string
   Return   : A(current time string)
   Format   : yy/mm/dd hh:mm:ss
              12345678901234567
  ---------------------------------------------------------*/
char *mkdate(char *yymmdd,int size) {
  time_t tloc;
  struct tm  *cur;

  if (size < SZ_DATE)   {
     memset(yymmdd,' ',size);
  }
  else {
     time(&tloc);
     cur = localtime(&tloc);
     strftime(yymmdd,size,"%y/%m/%d %T",cur);
  }
  return yymmdd;
}

void usage(char *prg,char *msg) {
  fprintf(stderr,"\nUsage : %s %s\n",prg,msg);
  fflush(stderr);
}

/*-----------------------------------------------------------------------
  Function : print error message
  -----------------------------------------------------------------------*/
FILE *log_fp = NULL;
  static char *err_msg[] = {
   "",
   "",
  };

char *fatal = "FATAL  ",
     *warn  = "WARNING";
void error(char *func,int no,char *info,short status) {
  char *mp;

  if (status)     mp = fatal;
  else            mp = warn;
  fprintf(stderr,"%s [%s] ",mp,func);fflush(stderr);

  if ((no >= ERR_UTL_S) && (no <= ERR_UTL_E))
    fprintf(stderr,"%s [%s]  err=%d\n",err_msg[no-ERR_UTL_S],info,no);
  else  {
    char *emsg;
    char buf[SZ_MSGBUF];
    emsg = (char *) strerror(errno);
    if (emsg)  strcpy(buf,emsg);
    else       buf[0] = '\0';
    strcat(buf," : ");
    strcat(buf,info);
    strcat(buf,"\n");
    fprintf(stderr,buf);
    if (log_fp)  {
      fprintf(log_fp,buf);
      fclose(log_fp);
    }
  }
  fflush(stderr);
  if (status)  exit(no);
  else         return;
}

 Makefile 

.IGNORE:
SHELL         = /bin/csh
CC            = gcc
RM            = rm -f

CFLAGS        = -c -g
CFLAGS_B      = -c -g -DBROADCAST -o
LDFLAGS       = -g -o
LDLIBS        = -lpthread

TARGET        = thdrun
TARGETB       = thdrunb
LIBS          = lib
SRCS          = $(TARGET).c $(LIBS).c
DEFS          = def.h lib.h
OBJS          = $(TARGET).o $(LIBS).o
OBJSB         = $(TARGETB).o $(LIBS).o

all:    $(TARGET)
all2:   $(TARGET) $(TARGETB)

$(TARGET):      $(OBJS)
                $(CC) $(LDFLAGS) $(TARGET) $(OBJS) $(LDLIBS)

$(TARGETB):     $(OBJSB)
                $(CC) $(LDFLAGS) $(TARGETB) $(OBJSB) $(LDLIBS)

$(TARGETB).o:   $(TARGET).c $(DEFS)
                $(CC) $(CFLAGS_B) $(TARGETB).o $(TARGET).c

$(TARGET).o:    $(TARGET).c $(DEFS)
$(LIBS).o:      $(LIBS).c   $(DEFS)

clean2:         clean cleanb
clean:
        $(RM) $(TARGET) $(OBJS) core $(TARGET).log

cleanb:
        $(RM) $(TARGETB) $(OBJSB) core $(TARGETB).log

スレッド・プログラミング