thread-pool.c 8.18 KB
Newer Older
laurent's avatar
laurent committed
1
/*
laurent's avatar
laurent committed
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21
* Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
* contributor license agreements.  See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The OpenAirInterface Software Alliance licenses this file to You under
* the OAI Public License, Version 1.1  (the "License"); you may not use this file
* except in compliance with the License.
* You may obtain a copy of the License at
*
*      http://www.openairinterface.org/?page_id=698
*
* Author and copyright: Laurent Thomas, open-cells.com
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*-------------------------------------------------------------------------------
* For more information about the OpenAirInterface (OAI) Software Alliance:
*      contact@openairinterface.org
laurent's avatar
laurent committed
22 23
*/

laurent's avatar
laurent committed
24

laurent's avatar
laurent committed
25 26 27 28 29 30 31
#define _GNU_SOURCE
#include <sched.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <string.h>
#include <unistd.h>
laurent's avatar
laurent committed
32 33 34
#include <ctype.h>
#include <sys/sysinfo.h>
#include <threadPool/thread-pool.h>
laurent's avatar
laurent committed
35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50

void displayList(notifiedFIFO_t *nf) {
  int n=0;
  notifiedFIFO_elt_t *ptr=nf->outF;

  while(ptr) {
    printf("element: %d, key: %lu\n",++n,ptr->key);
    ptr=ptr->next;
  }

  printf("End of list: %d elements\n",n);
}

static inline  notifiedFIFO_elt_t *pullNotifiedFifoRemember( notifiedFIFO_t *nf, struct one_thread *thr) {
  mutexlock(nf->lockF);

Robert Schmidt's avatar
Robert Schmidt committed
51
  while(!nf->outF && !thr->terminate)
laurent's avatar
laurent committed
52 53
    condwait(nf->notifF, nf->lockF);

Robert Schmidt's avatar
Robert Schmidt committed
54 55 56 57 58
  if (thr->terminate) {
    mutexunlock(nf->lockF);
    return NULL;
  }

laurent's avatar
laurent committed
59 60 61 62 63 64 65 66
  notifiedFIFO_elt_t *ret=nf->outF;
  nf->outF=nf->outF->next;

  if (nf->outF==NULL)
    nf->inF=NULL;

  // For abort feature
  thr->runningOnKey=ret->key;
Robert Schmidt's avatar
Robert Schmidt committed
67
  thr->dropJob = false;
laurent's avatar
laurent committed
68 69 70 71 72 73 74
  mutexunlock(nf->lockF);
  return ret;
}

void *one_thread(void *arg) {
  struct  one_thread *myThread=(struct  one_thread *) arg;
  struct  thread_pool *tp=myThread->pool;
laurent's avatar
laurent committed
75

laurent's avatar
laurent committed
76 77 78
  // Infinite loop to process requests
  do {
    notifiedFIFO_elt_t *elt=pullNotifiedFifoRemember(&tp->incomingFifo, myThread);
Robert Schmidt's avatar
Robert Schmidt committed
79 80 81 82
    if (elt == NULL) {
      AssertFatal(myThread->terminate, "pullNotifiedFifoRemember() returned NULL although thread not aborted\n");
      break;
    }
laurent's avatar
laurent committed
83

84
    if (tp->measurePerf) elt->startProcessingTime=rdtsc_oai();
laurent's avatar
laurent committed
85 86 87

    elt->processingFunc(NotifiedFifoData(elt));

88
    if (tp->measurePerf) elt->endProcessingTime=rdtsc_oai();
laurent's avatar
laurent committed
89 90 91 92 93

    if (elt->reponseFifo) {
      // Check if the job is still alive, else it has been aborted
      mutexlock(tp->incomingFifo.lockF);

Robert Schmidt's avatar
Robert Schmidt committed
94
      if (myThread->dropJob)
laurent's avatar
laurent committed
95 96 97
        delNotifiedFIFO_elt(elt);
      else
        pushNotifiedFIFO(elt->reponseFifo, elt);
98
      myThread->runningOnKey=-1;
laurent's avatar
laurent committed
99 100
      mutexunlock(tp->incomingFifo.lockF);
    }
Robert Schmidt's avatar
Robert Schmidt committed
101 102
  } while (!myThread->terminate);
  return NULL;
laurent's avatar
laurent committed
103 104
}

105
void initNamedTpool(char *params,tpool_t *pool, bool performanceMeas, char *name) {
laurent's avatar
laurent committed
106 107 108 109
  memset(pool,0,sizeof(*pool));
  char *measr=getenv("threadPoolMeasurements");
  pool->measurePerf=performanceMeas;
  // force measurement if the output is defined
110
  pool->measurePerf |= measr!=NULL;
laurent's avatar
laurent committed
111 112 113 114 115 116 117 118 119 120 121 122 123

  if (measr) {
    mkfifo(measr,0666);
    AssertFatal(-1 != (pool->dummyTraceFd=
                         open(measr, O_RDONLY| O_NONBLOCK)),"");
    AssertFatal(-1 != (pool->traceFd=
                         open(measr, O_WRONLY|O_APPEND|O_NOATIME|O_NONBLOCK)),"");
  } else
    pool->traceFd=-1;

  pool->activated=true;
  initNotifiedFIFO(&pool->incomingFifo);
  char *saveptr, * curptr;
Laurent's avatar
Laurent committed
124
  char *parms_cpy=strdup(params);
laurent's avatar
laurent committed
125
  pool->nbThreads=0;
Laurent's avatar
Laurent committed
126
  curptr=strtok_r(parms_cpy,",",&saveptr);
127
  struct one_thread * ptr;
128
  char *tname = (name == NULL ? "Tpool" : name);
laurent's avatar
laurent committed
129
  while ( curptr!=NULL ) {
laurent's avatar
laurent committed
130 131 132 133 134 135 136 137
    int c=toupper(curptr[0]);

    switch (c) {
      case 'N':
        pool->activated=false;
        break;

      default:
138
        ptr=pool->allthreads;
laurent's avatar
laurent committed
139
        pool->allthreads=(struct one_thread *)malloc(sizeof(struct one_thread));
140
        pool->allthreads->next=ptr;
laurent's avatar
laurent committed
141 142 143 144
        printf("create a thread for core %d\n", atoi(curptr));
        pool->allthreads->coreID=atoi(curptr);
        pool->allthreads->id=pool->nbThreads;
        pool->allthreads->pool=pool;
Robert Schmidt's avatar
Robert Schmidt committed
145 146
        pool->allthreads->dropJob = false;
        pool->allthreads->terminate = false;
147 148
        //Configure the thread scheduler policy for Linux
        // set the thread name for debugging
149
        sprintf(pool->allthreads->name,"%s%d_%d",tname,pool->nbThreads,pool->allthreads->coreID);
150 151
        threadCreate(&pool->allthreads->threadID, one_thread, (void *)pool->allthreads,
                     pool->allthreads->name, pool->allthreads->coreID, OAI_PRIORITY_RT);
laurent's avatar
laurent committed
152 153
        pool->nbThreads++;
    }
laurent's avatar
laurent committed
154 155 156

    curptr=strtok_r(NULL,",",&saveptr);
  }
Laurent's avatar
Laurent committed
157
  free(parms_cpy);
laurent's avatar
laurent committed
158 159 160 161 162 163 164
  if (pool->activated && pool->nbThreads==0) {
    printf("No servers created in the thread pool, exit\n");
    exit(1);
  }
}

#ifdef TEST_THREAD_POOL
165
volatile int oai_exit=0;
laurent's avatar
laurent committed
166

167 168 169
void exit_function(const char *file, const char *function, const int line, const char *s) {
}

laurent's avatar
laurent committed
170 171
struct testData {
  int id;
172
  int sleepTime;
laurent's avatar
laurent committed
173 174 175 176 177
  char txt[30];
};

void processing(void *arg) {
  struct testData *in=(struct testData *)arg;
178
  //printf("doing: %d, %s, in thr %ld\n",in->id, in->txt,pthread_self() );
laurent's avatar
laurent committed
179
  sprintf(in->txt,"Done by %ld, job %d", pthread_self(), in->id);
180 181 182
  in->sleepTime=rand()%1000;
  usleep(in->sleepTime);
  //printf("done: %d, %s, in thr %ld\n",in->id, in->txt,pthread_self() );
laurent's avatar
laurent committed
183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200
}

int main() {
  notifiedFIFO_t myFifo;
  initNotifiedFIFO(&myFifo);
  pushNotifiedFIFO(&myFifo,newNotifiedFIFO_elt(sizeof(struct testData), 1234,NULL,NULL));

  for(int i=10; i>1; i--) {
    pushNotifiedFIFO(&myFifo,newNotifiedFIFO_elt(sizeof(struct testData), 1000+i,NULL,NULL));
  }

  displayList(&myFifo);
  notifiedFIFO_elt_t *tmp=pullNotifiedFIFO(&myFifo);
  printf("pulled: %lu\n", tmp->key);
  displayList(&myFifo);
  tmp=pullNotifiedFIFO(&myFifo);
  printf("pulled: %lu\n", tmp->key);
  displayList(&myFifo);
201
  abortNotifiedFIFOJob(&myFifo,1005);
laurent's avatar
laurent committed
202 203 204 205
  printf("aborted 1005\n");
  displayList(&myFifo);
  pushNotifiedFIFO(&myFifo,newNotifiedFIFO_elt(sizeof(struct testData), 12345678, NULL, NULL));
  displayList(&myFifo);
206
  abortNotifiedFIFOJob(&myFifo,12345678);
laurent's avatar
laurent committed
207 208 209 210 211 212 213 214 215 216 217 218 219 220
  printf("aborted 12345678\n");
  displayList(&myFifo);

  do {
    tmp=pollNotifiedFIFO(&myFifo);

    if (tmp) {
      printf("pulled: %lu\n", tmp->key);
      displayList(&myFifo);
    } else
      printf("Empty list \n");
  } while(tmp);

  tpool_t  pool;
221
  char params[]="1,2,3,4,5";
laurent's avatar
laurent committed
222 223 224 225
  initTpool(params,&pool, true);
  notifiedFIFO_t worker_back;
  initNotifiedFIFO(&worker_back);

226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257
  sleep(1);
  int cumulProcessTime=0, cumulTime=0;
  struct timespec st,end;
  clock_gettime(CLOCK_MONOTONIC, &st);
  int nb_jobs=4;
  for (int i=0; i <1000 ; i++) {
    int parall=nb_jobs;
    for (int j=0; j <parall ; j++) {
      notifiedFIFO_elt_t *work=newNotifiedFIFO_elt(sizeof(struct testData), i, &worker_back, processing);
      struct testData *x=(struct testData *)NotifiedFifoData(work);
      x->id=i;
      pushTpool(&pool, work);
    }
    int sleepmax=0;
    while (parall) {
      tmp=pullTpool(&worker_back,&pool);
      if (tmp) {
	parall--;
	struct testData *dd=NotifiedFifoData(tmp);
	if (dd->sleepTime > sleepmax)
	  sleepmax=dd->sleepTime;
	delNotifiedFIFO_elt(tmp);
      }
    }
    cumulProcessTime+=sleepmax;
  }
  clock_gettime(CLOCK_MONOTONIC, &end);
  long long dur=(end.tv_sec-st.tv_sec)*1000*1000+(end.tv_nsec-st.tv_nsec)/1000;
  printf("In µs, Total time per group of %d job:%lld, work time per job %d, overhead per job %lld\n",
	 nb_jobs, dur/1000, cumulProcessTime/1000, (dur-cumulProcessTime)/(1000*nb_jobs));

	/*	
laurent's avatar
laurent committed
258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274
  for (int i=0; i <1000 ; i++) {
    notifiedFIFO_elt_t *work=newNotifiedFIFO_elt(sizeof(struct testData), i, &worker_back, processing);
    struct testData *x=(struct testData *)NotifiedFifoData(work);
    x->id=i;
    pushTpool(&pool, work);
  }

  do {
    tmp=pullTpool(&worker_back,&pool);

    if (tmp) {
      struct testData *dd=NotifiedFifoData(tmp);
      printf("Result: %s\n",dd->txt);
      delNotifiedFIFO_elt(tmp);
    } else
      printf("Empty list \n");

275
    abortTpoolJob(&pool,510);
laurent's avatar
laurent committed
276
  } while(tmp);
277
	*/
laurent's avatar
laurent committed
278 279 280
  return 0;
}
#endif