thread-pool.c 14 KB
Newer Older
laurent's avatar
laurent committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40
#define __USE_GNU
#define _GNU_SOURCE
#include <stdio.h>
#include <pthread.h>
#include <sched.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <string.h>
#include <unistd.h>
#include <sys/syscall.h>
#include <sys/time.h>
#include <stdbool.h>
// OAI includes
#include <assertions.h>
#include <log.h>
#include "PHY/TOOLS/time_meas.h"
#include "PHY/CODING/defs.h"
#include "PHY/CODING/extern.h"
#include <thread-pool.h>

#ifdef DEBUG
#define THREADINIT   PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP
#else
#define THREADINIT   PTHREAD_MUTEX_INITIALIZER
#endif


request_t * createRequest(enum request_t type,int size) {
    request_t* request;
    AssertFatal( (request = (request_t*)aligned_alloc(32,sizeof(request_t)+size)) != NULL,"");
    request->id = 0;
    request->type=type;
    request->next = NULL;
    request->creationTime=rdtsc();
    request->data=(void*)(request+1);
    return request;
}

void freeRequest(request_t* request) {
laurent's avatar
laurent committed
41
    //printf("freeing: %ld, %p\n", request->id, request);
laurent's avatar
laurent committed
42 43 44
    free(request);
}

laurent's avatar
laurent committed
45
volatile int ii=0;
laurent's avatar
laurent committed
46
int add_request(request_t* request, tpool_t * tp) {
laurent's avatar
laurent committed
47
    mutexlock(tp->lockRequests);
laurent's avatar
laurent committed
48 49 50 51 52 53 54
    if (tp->oldestRequests == NULL)
        tp->oldestRequests = request;
    else {
        AssertFatal(tp->newestRequests != NULL, "");
        tp->newestRequests->next = request;
    }
    tp->newestRequests = request;
laurent's avatar
laurent committed
55
    mutexlock(tp->lockReportDone);
laurent's avatar
laurent committed
56
    tp->notFinishedJobs++;
laurent's avatar
laurent committed
57 58 59
    mutexunlock(tp->lockReportDone);
    condbroadcast(tp->notifRequest);
    mutexunlock(tp->lockRequests);
laurent's avatar
laurent committed
60 61 62 63 64 65
    return 0;
}

int add_requests(uint64_t request_num, tpool_t * tp) {
    request_t* request;
    int nbToAdd=((uint32_t)lrand48())%20+1;
laurent's avatar
laurent committed
66
    mutexlock(tp->lockRequests);
laurent's avatar
laurent committed
67 68 69
    for (int i=0; i<nbToAdd; i++) {
        // simulate request
        request=createRequest(DECODE,sizeof(turboDecode_t));
laurent's avatar
laurent committed
70
        union turboReqUnion id= {.s={request_num,1000,i*10,111,222}};
laurent's avatar
laurent committed
71 72 73 74 75 76 77 78 79 80 81 82 83 84
        request->id= id.p;
        turboDecode_t * rdata=(turboDecode_t *) request->data;
        rdata->function=phy_threegpplte_turbo_decoder8;
        rdata->Kr=6144;
        rdata->iind=0; // not used, OAI code need cleanup!!!
        rdata->Fbits=0;
        rdata->maxIterations=6;
        if (tp->oldestRequests == NULL)
            tp->oldestRequests = request;
        else
            tp->newestRequests->next = request;
        tp->newestRequests = request;
    }

laurent's avatar
laurent committed
85 86 87 88 89
    mutexlock(tp->lockReportDone);
    tp->notFinishedJobs+=nbToAdd;
    mutexunlock(tp->lockReportDone);
    condbroadcast(tp->notifRequest);
    mutexunlock(tp->lockRequests);
laurent's avatar
laurent committed
90 91 92 93
    return nbToAdd;
}

request_t * get_request(tpool_t * tp, uint16_t threadID ) {
laurent's avatar
laurent committed
94 95 96 97 98 99
    int nb=0;
    request_t* r=tp->oldestRequests;
    while (r!=NULL) {
        nb++;
        r=r->next;
    }
laurent's avatar
laurent committed
100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123
    request_t* request=tp->oldestRequests;
    if (request == NULL)
        return NULL;

    if ( tp->restrictRNTI ) {
        request_t** start=&tp->oldestRequests;
        request = NULL;
        while(*start!=NULL && request==NULL) {
            union turboReqUnion id= {.p=(*start)->id};
            if ( id.s.rnti % tp->nbThreads ==  threadID ) {
                request=*start;
                *start=(*start)->next;
            } else
                start=&((*start)->next);
        }
    } else
        tp->oldestRequests = request->next;

    if ( tp->oldestRequests == NULL)
        tp->newestRequests=NULL;

    int nnb=0;
    r=tp->oldestRequests;
    while (r!=NULL) {
laurent's avatar
laurent committed
124 125
        nnb++;
        r=r->next;
laurent's avatar
laurent committed
126
    }
laurent's avatar
laurent committed
127 128 129 130
    /*
    if ( ! ( nb == nnb && request == NULL))
      printf("getr:was=%d,is=%d,gotit=%p\n",nb,nnb,request);
    */
laurent's avatar
laurent committed
131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152
    return request;
}

request_t * searchRNTI(tpool_t * tp, rnti_t rnti) {
    request_t * result=NULL;
    request_t ** start=&tp->oldestRequests;
    while(*start!=NULL && result==NULL) {
        union turboReqUnion id= {.p=(*start)->id};
        if ( id.s.rnti == rnti ) {
            result=*start;
            *start=(*start)->next;
            if ( tp->oldestRequests == NULL)
                tp->newestRequests=NULL;
        } else
            start=&((*start)->next);
    }
    return result;
}



void process_request(request_t* request) {
laurent's avatar
laurent committed
153
    //printf("S:%s...",request->type==DECODE?"D":"E");
laurent's avatar
laurent committed
154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193
    switch(request->type) {
    case DECODE : {
        time_stats_t oaitimes[7];
        turboDecode_t * rdata=(turboDecode_t*) request->data;
        rdata->decodeIterations=rdata->function(rdata->soft_bits+96,
                                                rdata->decoded_bytes,
                                                rdata->Kr,
                                                f1f2mat_old[rdata->iind*2],
                                                f1f2mat_old[(rdata->iind*2)+1],
                                                rdata->maxIterations,
                                                rdata->nbSegments == 1 ? CRC24_A: CRC24_B,
                                                rdata->Fbits,
                                                oaitimes+0,
                                                oaitimes+1,
                                                oaitimes+2,
                                                oaitimes+3,
                                                oaitimes+4,
                                                oaitimes+5,
                                                oaitimes+6);
    };
    break;
    case ENCODE :  {
        turboEncode_t * rdata=(turboEncode_t*) request->data;
        memset(rdata->output,LTE_NULL,TURBO_SIMD_SOFTBITS);
        threegpplte_turbo_encoder(rdata->input,
                                  rdata->Kr_bytes,
                                  rdata->output+96,//&dlsch->harq_processes[harq_pid]->d[r][96],
                                  rdata->filler,
                                  f1f2mat_old[rdata->iind*2],   // f1 (see 36121-820, page 14)
                                  f1f2mat_old[(rdata->iind*2)+1]  // f2 (see 36121-820, page 14)
                                 );
    };
    break;
    default:
        AssertFatal(false,"");
    }
    //printf("..End\n");
}

void handle_request(tpool_t * tp, request_t* request) {
laurent's avatar
laurent committed
194 195 196 197 198 199 200 201 202 203 204 205 206 207 208
    request->startProcessingTime=rdtsc();
    process_request(request);
    request->endProcessingTime=rdtsc();
    mutexlock(tp->lockReportDone);
    tp->notFinishedJobs--;
    request->next=tp->doneRequests;
    tp->doneRequests=request;
    condsignal(tp->notifDone);
    mutexunlock(tp->lockReportDone);
    /*
      printf("Thread '%ld' handled request '%d' delay in µs:%ld\n",
      syscall( SYS_gettid ),
      request->id,
      (rdtsc() - request->creationTime)/tp->cpuCyclesMicroSec);
    */
laurent's avatar
laurent committed
209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232
}

void* one_thread(void* data) {
    struct  one_thread * myThread=(struct  one_thread *) data;
    struct  thread_pool* tp=myThread->pool;

    // configure the thread core assignment
    // TBD: reserve the core for us exclusively
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(myThread->coreID, &cpuset);
    pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);

    //Configure the thread scheduler policy for Linux
    struct sched_param sparam= {0};
    sparam.sched_priority = sched_get_priority_max(SCHED_RR);
    pthread_setschedparam(pthread_self(), SCHED_RR, &sparam);

    // set the thread name for debugging
    sprintf(myThread->name,"Tcodec_%d",myThread->coreID);
    pthread_setname_np(pthread_self(), myThread->name );

    // Infinite loop to process requests
    do {
laurent's avatar
laurent committed
233 234
        mutexlock(tp->lockRequests);

laurent's avatar
laurent committed
235 236
        request_t* request = get_request(tp, myThread->id);
        if (request == NULL) {
laurent's avatar
laurent committed
237 238 239 240 241 242
            condwait(tp->notifRequest,tp->lockRequests);
            request = get_request(tp, myThread->id);
        }

        mutexunlock(tp->lockRequests);

laurent's avatar
laurent committed
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
        if (request!=NULL) {
            strncpy(request->processedBy,myThread->name, 15);
            request->coreId=myThread->coreID;
            handle_request(tp, request);
        }

    } while (true);
}

void init_tpool(char * params,tpool_t * pool) {
    mkfifo("/tmp/test-tcri",0666);
    pool->dummyTraceFd=open("/tmp/test-tcri", O_RDONLY| O_NONBLOCK);
    if ( pool->dummyTraceFd == -1 ) {
        perror("open read mode trace file:");
        exit(1);
    }
    pool->traceFd=open("/tmp/test-tcri", O_WRONLY|O_APPEND|O_NOATIME|O_NONBLOCK);
    if ( pool->traceFd == -1 ) {
        perror("open trace file:");
        exit(1);
    }

    //Configure the thread scheduler policy for Linux
    struct sched_param sparam= {0};
    sparam.sched_priority = sched_get_priority_max(SCHED_RR)-1;
    pthread_setschedparam(pthread_self(), SCHED_RR, &sparam);
    pool->activated=true;
laurent's avatar
laurent committed
270 271 272 273 274
    mutexinit(pool->lockRequests);
    condinit (pool->notifRequest);
    pool->notifCount=0;
    mutexinit(pool->lockReportDone);
    condinit (pool->notifDone);
laurent's avatar
laurent committed
275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362
    pool->oldestRequests=NULL;
    pool->newestRequests=NULL;
    pool->doneRequests=NULL;
    pool->notFinishedJobs=0;
    pool->allthreads=NULL;
    char * saveptr, * curptr;
    pool->nbThreads=0;
    pool->restrictRNTI=false;
    curptr=strtok_r(params,",",&saveptr);
    while ( curptr!=NULL ) {
        if (curptr[0] == 'u' || curptr[0] == 'U') {
            pool->restrictRNTI=true;
        } else if ( curptr[0]>='0' && curptr[0]<='9' ) {
            struct one_thread *tmp=pool->allthreads;
            pool->allthreads=(struct one_thread *)malloc(sizeof(struct one_thread));
            pool->allthreads->next=tmp;
            printf("create a thread for core %d\n", atoi(curptr));
            pool->allthreads->coreID=atoi(curptr);
            pool->allthreads->id=pool->nbThreads;
            pool->allthreads->pool=pool;
            pthread_create(&pool->allthreads->threadID, NULL, one_thread, (void*)pool->allthreads);
            pool->nbThreads++;
        } else if (curptr[0] == 'n' || curptr[0] == 'N') {
            pool->activated=false;
        } else
            printf("Error in options for thread pool: %s\n",curptr);
        curptr=strtok_r(NULL,",",&saveptr);
    }

    if (pool->activated && pool->nbThreads==0) {
        printf("No servers created in the thread pool, exit\n");
        exit(1);
    }

    uint64_t deb=rdtsc();
    usleep(100000);
    pool->cpuCyclesMicroSec=(rdtsc()-deb)/100000;
    printf("Cycles per µs: %lu\n",pool->cpuCyclesMicroSec);

}


void displayList(request_t*start, request_t*end) {
    int n=0;
    while(start!=NULL) {
        n++;
        union turboReqUnion id= {.p=start->id};
        printf("rnti:%u frame:%u-%u codeblock:%u\n",
               id.s.rnti,
               id.s.frame,
               id.s.subframe,
               id.s.codeblock);
        if ( start->next==NULL && start!=end)
            printf("Error of end pointer");
        start=start->next;
    }
    printf("End of list: %d elements\n",n);
}

#ifdef TESTMAIN
#include "PHY/CODING/lte_interleaver.h"
#include "PHY/CODING/lte_interleaver2.h"

int main(int argc, char* argv[]) {

    if (argc<2) {
        printf("Usage: %s core,core,...\n",argv[0]);
        exit(1);
    }

    // configure the thread core assignment: client thread on core 0
    // TBD: reserve the core for us exclusively
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(0, &cpuset);
    pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
    tpool_t  pool;
    init_tpool(argv[1], &pool);
    //initialize turbo decoder tables
    init_td8();

    uint64_t i=1;
    // Test the lists
    srand48(time(NULL));
    int nbRequest=add_requests(i, &pool);
    printf("These should be: %d elements in the list\n",nbRequest);
    displayList(pool.oldestRequests, pool.newestRequests);
    // remove in middle
laurent's avatar
laurent committed
363
    request_t *req106=searchRNTI(&pool, 106);
laurent's avatar
laurent committed
364 365 366 367 368 369 370 371 372 373 374
    if (req106) {
        union turboReqUnion id= {.p=req106->id};
        printf("Removed: rnti:%u frame:%u-%u codeblock:%u, check it\n",
               id.s.rnti,
               id.s.frame,
               id.s.subframe,
               id.s.codeblock);
        freeRequest(req106);
    }  else
        printf("no rnti 106\n");
    displayList(pool.oldestRequests, pool.newestRequests);
laurent's avatar
laurent committed
375
    request_t *reqlast=searchRNTI(&pool, 100+nbRequest-1);
laurent's avatar
laurent committed
376 377 378 379 380 381 382 383
    if (reqlast) {
        printf("Removed last item, check it\n");
        freeRequest(reqlast);
    }  else
        printf("tried to removed from empty list\n");
    displayList(pool.oldestRequests, pool.newestRequests);
    printf("Remove all jobs\n");
    while(pool.oldestRequests!=NULL)
laurent's avatar
laurent committed
384
        get_request(&pool,0);
laurent's avatar
laurent committed
385 386
    printf("List should be empty now\n");
    displayList(pool.oldestRequests, pool.newestRequests);
laurent's avatar
laurent committed
387 388 389 390 391 392

    sleep(1);
    mutexlock(pool.lockReportDone);
    pool.notFinishedJobs=0;
    pool.doneRequests=NULL;
    mutexunlock(pool.lockReportDone);
laurent's avatar
laurent committed
393 394 395 396

    while (1) {
        uint64_t now=rdtsc();
        /* run a loop that generates a lot of requests */
laurent's avatar
laurent committed
397 398 399 400
        AssertFatal(pool.notFinishedJobs==0,"");
        int n=add_requests(i, &pool);
        printf("Added %d requests\n",n);

laurent's avatar
laurent committed
401 402
        /*
            // The main thread also process the queue
laurent's avatar
laurent committed
403
            mutexlock(pool.lockRequests);
laurent's avatar
laurent committed
404
            request_t* request= NULL;
laurent's avatar
laurent committed
405 406
            while ( (request=get_request(&pool,0)) != NULL ) {
                mutexunlock(pool.lockRequests);
laurent's avatar
laurent committed
407 408
                strcpy(request->processedBy,"MainThread");
                handle_request(&pool, request);
laurent's avatar
laurent committed
409
                mutexlock(pool.lockRequests);
laurent's avatar
laurent committed
410
            }
laurent's avatar
laurent committed
411
            mutexunlock(pool.lockRequests);
laurent's avatar
laurent committed
412 413 414
        */

        // Wait all other threads finish to process
laurent's avatar
laurent committed
415
        mutexlock(pool.lockReportDone);
laurent's avatar
laurent committed
416
        while ( pool.notFinishedJobs > 0 ) {
laurent's avatar
laurent committed
417
            condwait(pool.notifDone,pool.lockReportDone);
laurent's avatar
laurent committed
418
        }
laurent's avatar
laurent committed
419 420 421 422 423 424 425 426
        mutexunlock(pool.lockReportDone);

        int i=0;
        for (request_t* ptr=pool.doneRequests; ptr!=NULL; ptr=ptr->next)  {
            i++;
            //printf("in return: %ld, %p\n", ptr->id, ptr);
        }
        AssertFatal(i==n,"%d/%d\n",i,n);
laurent's avatar
laurent committed
427 428 429

        while (pool.doneRequests!=NULL) {
            pool.doneRequests->returnTime=rdtsc();
laurent's avatar
laurent committed
430
            if(write(pool.traceFd,pool.doneRequests,sizeof(request_t)- 2*sizeof(void*))) {};
laurent's avatar
laurent committed
431 432
            request_t* tmp=pool.doneRequests;
            pool.doneRequests=pool.doneRequests->next;
laurent's avatar
laurent committed
433
            freeRequest(tmp);
laurent's avatar
laurent committed
434
        }
laurent's avatar
laurent committed
435 436

        printf("Requests %d Done in %ld µsec\n",i, (rdtsc()-now)/pool.cpuCyclesMicroSec);
laurent's avatar
laurent committed
437 438 439 440 441
        i++;
    };
    return 0;
}
#endif