52matlab技术网站，matlab教程，matlab安装教程，matlab下载

标题: linux 多线程处理 [打印本页]

作者: hjman 时间: 2015-3-12 11:54
标题: linux 多线程处理
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <pthread.h>
#include <assert.h>
#include <time.h>
#include <unistd.h>
#include <curses.h>

#define MAX_SIZE 10
#define WORK_NUMBER 100
#define MAX_THREADS 5

int id_num = 0;
int id = 1;

int exitthread = 0;

typedef struct worker {

void *(*process)(void *arg);
void *arg;
int id;
struct worker *prev;
struct worker *next;
int data;
int flag ;

} CalThread_worker;

typedef struct {
pthread_mutex_t queue_lock;
pthread_cond_t queue_ready;

CalThread_worker *queue_head;

int shutdown;
pthread_t *threadid;

int max_thread_num;

int cur_queue_size;

int cal_counts;

int r[MAX_SIZE];

int addid[MAX_SIZE];

int subid[MAX_SIZE];

int addflag;

int subflag;

int start;

int sumdata;

} CalThread_pool;

int pool_add_worker(void *(*process)(void *arg), void *arg);
void *thread_routine(void *arg);
void *thread_moniter(void *arg);

//share resource
static CalThread_pool *pool = NULL;
void pool_init(int max_thread_num)
{
int i = 0;
pthread_t moniter_tid;
srand(clock());
pool = (CalThread_pool *) malloc(sizeof(CalThread_pool));
pthread_mutex_init(&(pool->queue_lock), NULL);
pthread_cond_init(&(pool->queue_ready), NULL);
pool->queue_head = NULL;
pool->max_thread_num = max_thread_num;
pool->cur_queue_size = 0;
pool->shutdown = 0;
pool->start = 0;
pool->cal_counts = 0;
//锟斤拷锟揭?拷锟斤拷锟侥诧拷锟斤拷
for (i = 0; i < MAX_SIZE; i++) {
      pool->addid[i] = rand() % 100;
      pool->subid[i] = rand() % 100;
      pool->r[i] = rand() & 1000;
}
pool->subflag = 1;
printf("%d,%d\n", pool->addid, pool->subid);
//}
pool->threadid = (pthread_t *) malloc(max_thread_num * sizeof(pthread_t));
for (i = 0; i < max_thread_num; i++) {
      pthread_create (&(pool->threadid[i]), NULL, thread_routine,NULL);
}

pthread_create (&moniter_tid, NULL, thread_moniter,NULL);
}

int pool_add_worker(void *(*process)(void *arg), void *arg)
{
CalThread_worker *newworker = (CalThread_worker *) malloc(
                                    sizeof(CalThread_worker));
newworker->process = process;
newworker->arg = arg;
newworker->next = NULL;
pthread_mutex_lock(&(pool->queue_lock));
CalThread_worker *member = pool->queue_head;
if (member != NULL) {
      while (member->next != NULL)
         member = member->next;
      member->next = newworker;
      newworker->prev = member;
} else {
      pool->queue_head = newworker;
      newworker->prev = NULL;
}
newworker->id = id_num++;
newworker->data = 1000;

//assert (pool->queue_head != NULL);
pool->cur_queue_size++;
pthread_mutex_unlock(&(pool->queue_lock));

return 0;
}

int pool_destroy()
{
if (pool->shutdown)
      return -1;
pool->start = 0;
pool->cal_counts = 0;
pool->shutdown = 1;
pthread_cond_broadcast(&(pool->queue_ready));
int i;
for (i = 0; i < pool->max_thread_num; i++)
      pthread_join(pool->threadid[i], NULL);
free(pool->threadid);
CalThread_worker *head = NULL;
while (pool->queue_head != NULL) {
      head = pool->queue_head;
      pool->queue_head = pool->queue_head->next;
      free(head);
}
pthread_mutex_destroy(&(pool->queue_lock));
pthread_cond_destroy(&(pool->queue_ready));
free(pool);
pool = NULL;
printf("pool_destroy\n");
return 0;
}

void * thread_routine(void *arg)
{
//printf ("starting thread 0x%x\n", pthread_self ());
while (1) {
      pthread_mutex_lock(&(pool->queue_lock));
      while (pool->cal_counts == 0 && !pool->shutdown) {
         printf("thread 0x%x is waiting\n", pthread_self());
         pthread_cond_wait(&(pool->queue_ready), &(pool->queue_lock));
      }
      if (pool->shutdown) {
         pthread_mutex_unlock(&(pool->queue_lock));
         printf("thread 0x%x will exit\n", pthread_self());
         pthread_exit(NULL);
      }
      printf("thread 0x%x is starting to work\n", pthread_self());
      //assert (pool->queue_head != NULL);
      //pool->cur_queue_size--;
      usleep(1000*500);
      CalThread_worker *worker = NULL;
      if (pool->start == 1) {
         worker = pool->queue_head;
         if (pool->subflag) {
            do {
                  if (worker->id == pool->subid[pool->cal_counts-1])
                     break;
                  worker = worker->next;
            } while (worker != NULL);
            worker->flag  = 1;
            pool->addflag = 1;
            pool->subflag = 0;
         } else if (pool->addflag) {
            do {
                  if (worker->id == pool->addid[pool->cal_counts-1])
                     break;
                  worker = worker->next;
            } while (worker != NULL);
            if (worker != NULL) {
                  pool->addflag = 0;
                  pool->subflag = 1;
                  worker->flag  = 2;
            }
         }
      }
      pthread_mutex_unlock(&(pool->queue_lock));
      if (worker != NULL) {
         (*(worker->process))(worker->arg);
      }
}
pthread_exit (NULL);
return 0;
}

void * thread_moniter(void *arg){

while (1) {
   char ch = getchar();
   if (ch == 'e' || ch == 'E')break;
   usleep(1000*500);
}
exitthread = 1;
pthread_exit (NULL);

return NULL;
}

void * calprocess(void *arg)
{
int i ;
//printf("threadid is 0x%x, working on task %d\n", pthread_self(), *(int *) arg);
CalThread_worker *worker = NULL;
worker = pool->queue_head;
int orgdata = 0;
do {
      if (worker->id == *(int *) arg)
         break;
      worker = worker->next;
} while (worker != NULL);
if (worker == NULL)return NULL;
if (worker->flag == 1) {
      orgdata = worker->data ;
      worker->data -= pool->r[pool->cal_counts -1];
      pool->sumdata = worker->data;
      printf ("the %d calcuate \n",id++);
      printf ("threadid :0x%x, sub-data:%d\n",pthread_self(),worker->data );
      worker->data = orgdata;
}
if (worker->flag == 2) {
      orgdata = worker->data;
      worker->data += pool->r[pool->cal_counts -1];
pool->cal_counts--;
      if (pool->cal_counts == 0){
pool->start = 0;
for (i = 0; i < MAX_SIZE; i++) {
            pool->addid[i] = rand() % 100;
            pool->subid[i] = rand() % 100;
            pool->r[i] = rand() & 1000;
         }
pool->start = 1;
pool->cal_counts = MAX_SIZE;
      }
      pool->sumdata += worker->data;
      printf ("threadid :0x%x, add-data:%d\n",pthread_self(),worker->data );
      printf ("threadid :0x%x, sum-data:%d\n",pthread_self(),pool->sumdata);
      worker->data = orgdata;
}
worker->flag = 0;
return NULL;
}

int main(int argc, char **argv)
{
int i = 0;
pool_init(MAX_THREADS);
pool->subflag = 1;
pool->start = 1;
int *workingnum = (int *) malloc(sizeof(int) * WORK_NUMBER);

for (i = 0; i < WORK_NUMBER; i++) {
      workingnum[i] = i;
      pool_add_worker(calprocess, &workingnum[i]);
}
pool->cal_counts = MAX_SIZE;
printf("counts:%d\n", pool->cal_counts);

pthread_cond_broadcast(&(pool->queue_ready));
while (!exitthread)
   sleep(2);

pool_destroy();
free(workingnum);
printf("main end\n");
return 0;

欢迎光临 52matlab技术网站，matlab教程，matlab安装教程，matlab下载 (http://www.52matlab.com/)