thread-pool.c 8.47 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);
    }
101 102
    else
      delNotifiedFIFO_elt(elt);
Robert Schmidt's avatar
Robert Schmidt committed
103 104
  } while (!myThread->terminate);
  return NULL;
laurent's avatar
laurent committed
105 106
}

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

  if (measr) {
    mkfifo(measr,0666);
116
    AssertFatal(-1 != (pool->dummyKeepReadingTraceFd=
laurent's avatar
laurent committed
117 118 119 120 121 122 123 124 125
                         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
126
  char *parms_cpy=strdup(params);
laurent's avatar
laurent committed
127
  pool->nbThreads=0;
Laurent's avatar
Laurent committed
128
  curptr=strtok_r(parms_cpy,",",&saveptr);
129
  struct one_thread * ptr;
130
  char *tname = (name == NULL ? "Tpool" : name);
laurent's avatar
laurent committed
131
  while ( curptr!=NULL ) {
laurent's avatar
laurent committed
132 133 134
    int c=toupper(curptr[0]);

    switch (c) {
135

laurent's avatar
laurent committed
136 137 138 139 140
      case 'N':
        pool->activated=false;
        break;

      default:
141
        ptr=pool->allthreads;
laurent's avatar
laurent committed
142
        pool->allthreads=(struct one_thread *)malloc(sizeof(struct one_thread));
143
        pool->allthreads->next=ptr;
laurent's avatar
laurent committed
144 145 146 147
        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
148 149
        pool->allthreads->dropJob = false;
        pool->allthreads->terminate = false;
150 151
        //Configure the thread scheduler policy for Linux
        // set the thread name for debugging
152
        sprintf(pool->allthreads->name,"%s%d_%d",tname,pool->nbThreads,pool->allthreads->coreID);
153 154
        threadCreate(&pool->allthreads->threadID, one_thread, (void *)pool->allthreads,
                     pool->allthreads->name, pool->allthreads->coreID, OAI_PRIORITY_RT);
laurent's avatar
laurent committed
155 156
        pool->nbThreads++;
    }
laurent's avatar
laurent committed
157 158 159

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

167 168 169 170 171
void initFloatingCoresTpool(int nbThreads,tpool_t *pool, bool performanceMeas, char *name) {
  char threads[1024] = "n";
  if (nbThreads) {
    strcpy(threads,"-1");
    for (int i=1; i < nbThreads; i++)
172
      strncat(threads,",-1", sizeof(threads)-1);
173
  }
174
  threads[sizeof(threads)-1]=0;
175 176 177
  initNamedTpool(threads, pool, performanceMeas, name);
}

laurent's avatar
laurent committed
178
#ifdef TEST_THREAD_POOL
179
volatile int oai_exit=0;
laurent's avatar
laurent committed
180

181 182 183 184 185 186 187
void exit_function(const char *file, const char *function, const int line, const char *s, const int assert)
{
  if (assert) {
    abort();
  } else {
    exit(EXIT_SUCCESS);
  }
188 189
}

laurent's avatar
laurent committed
190 191
struct testData {
  int id;
192
  int sleepTime;
laurent's avatar
laurent committed
193 194 195 196 197
  char txt[30];
};

void processing(void *arg) {
  struct testData *in=(struct testData *)arg;
198
  //printf("doing: %d, %s, in thr %ld\n",in->id, in->txt,pthread_self() );
laurent's avatar
laurent committed
199
  sprintf(in->txt,"Done by %ld, job %d", pthread_self(), in->id);
200 201 202
  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
203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234
}

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);
  pushNotifiedFIFO(&myFifo,newNotifiedFIFO_elt(sizeof(struct testData), 12345678, NULL, NULL));
  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;
235
  char params[]="1,2,3,4,5";
laurent's avatar
laurent committed
236 237 238 239
  initTpool(params,&pool, true);
  notifiedFIFO_t worker_back;
  initNotifiedFIFO(&worker_back);

240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271
  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
272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288
  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");

289
    abortTpoolJob(&pool,510);
laurent's avatar
laurent committed
290
  } while(tmp);
291
	*/
laurent's avatar
laurent committed
292 293 294
  return 0;
}
#endif