examples/xbt/sem_sched.c

   1 /* Copyright (c) 2007, 2008. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include <stdio.h>
   8 #include <stdlib.h>
   9 #include <string.h>
  10 #include <errno.h>
  11
  12
  13 #include "xbt/xbt_os_thread.h"
  14 #include "xbt.h"
  15 #include "xbt/log.h"
  16 XBT_LOG_NEW_DEFAULT_CATEGORY(sem_sched,"Messages specific for this sem example");
  17
  18 #ifndef ENOBUFS
  19 #define ENOBUFS         1024
  20 #endif
  21
  22 #define CTX_MAX                 ((unsigned int)1000)
  23
  24
  25 #define MAX_ARG                                         30
  26 #define MAX_ARGS                                        10
  27
  28 typedef int (*pfn_func_t)(int, char**);
  29
  30 static int
  31 __next_ctx_ID = 0;
  32
  33 typedef struct s_job
  34 {
  35         pfn_func_t func;
  36         int argc;
  37         char** argv;
  38 }s_job_t,* job_t;
  39
  40
  41 job_t
  42 job_new(pfn_func_t func, int argc, char** argv);
  43
  44 int
  45 job_execute(job_t job);
  46
  47 int
  48 job_free(job_t* ref);
  49
  50 /* an entry of the table of threads */
  51 typedef struct s_ctx
  52 {
  53         xbt_os_sem_t begin;
  54         xbt_os_sem_t end;
  55         int failure;
  56         job_t job;
  57         xbt_os_thread_t imp;
  58         int index;
  59 }s_ctx_t,* ctx_t;
  60
  61 typedef struct s_shed
  62 {
  63         ctx_t* ctxs;
  64         int size;
  65         int capacity;
  66 }s_sched_t,* sched_t;
  67
  68
  69 void
  70 schedule(ctx_t c);
  71 void
  72 unschedule(ctx_t c);
  73
  74 void*
  75 ctx_function(void* param);
  76
  77 ctx_t
  78 ctx_new(job_t job);
  79
  80 int
  81 ctx_free(ctx_t* ref);
  82
  83 sched_t
  84 sched_new(int size);
  85
  86 int
  87 sched_add_job(sched_t sched, job_t job);
  88
  89 int
  90 sched_init(sched_t sched);
  91
  92 int
  93 sched_schedule(sched_t sched);
  94
  95 int
  96 sched_clean(sched_t sched);
  97
  98 int
  99 sched_free(sched_t* ref);
 100
 101 int
 102 job(int argc, char* argv[]);
 103
 104 int
 105 main(int argc, char* argv[])
 106 {
 107         sched_t sched;
 108         int i, size;
 109         char** __argv;
 110         char arg[MAX_ARG] = {0};
 111
 112
 113         xbt_init(&argc, argv);
 114
 115         if(argc != 2)
 116         {
 117                 INFO1("Usage: %s job count",argv[0]);
 118                 exit(EXIT_FAILURE);
 119         }
 120
 121
 122         size = atoi(argv[1]);
 123
 124         /* create a new scheduler */
 125         sched = sched_new(size);
 126
 127         if(!sched)
 128         {
 129                 INFO1("sched_new() failed : errno %d",errno);
 130                 exit(EXIT_FAILURE);
 131         }
 132
 133         __argv = xbt_new0(char*,MAX_ARGS);
 134
 135         for(i = 0; i < MAX_ARGS; i++)
 136         {
 137                 sprintf(arg,"arg_%d",i);
 138                 __argv[i] = strdup(arg);
 139                 memset(arg,0,MAX_ARG);
 140
 141         }
 142
 143         for(i = 0; i < size; i++)
 144                 sched_add_job(sched,job_new(job,(i < MAX_ARGS) ? i : MAX_ARGS,__argv));
 145
 146         /* initialize the scheduler */
 147         if(sched_init(sched) < 0)
 148         {
 149                 sched_free(&sched);
 150                 INFO1("sched_init() failed : errno %d\n",errno);
 151                 exit(EXIT_FAILURE);
 152         }
 153
 154         /* schedule the jobs */
 155         if(sched_schedule(sched) < 0)
 156         {
 157                 sched_free(&sched);
 158                 INFO1("sched_init() failed : errno %d",errno);
 159                 exit(EXIT_FAILURE);
 160         }
 161
 162         /* cleanup */
 163         if(sched_clean(sched) < 0)
 164         {
 165                 sched_free(&sched);
 166                 INFO1("sched_init() failed : errno %d",errno);
 167                 exit(EXIT_FAILURE);
 168         }
 169
 170         /* destroy the scheduler */
 171         sched_free(&sched);
 172
 173         INFO1("sem_sched terminated with exit code %d (success)",EXIT_SUCCESS);
 174
 175         return EXIT_SUCCESS;
 176
 177 }
 178
 179 void*
 180 ctx_function(void* param)
 181 {
 182         int i = 0;
 183         int exit_code = 1;
 184         ctx_t ctx = (ctx_t)param;
 185
 186         INFO1("Hello i'm the owner of the context %d, i'm waiting for starting",ctx->index);
 187
 188         unschedule(ctx);
 189
 190         if(ctx->failure)
 191         {
 192                 INFO1("0ups the scheduler initialization failed bye {%d}.",ctx->index);
 193                 xbt_os_thread_exit(&exit_code);
 194         }
 195
 196         INFO1("I'm the owner of the context %d : I'm started",ctx->index);
 197         INFO0("Wait a minute, I do my job");
 198
 199         /* do its job */
 200         exit_code = job_execute(ctx->job);
 201
 202         INFO1("Have finished my job, bye {%d}\n",ctx->index);
 203
 204         xbt_os_sem_release(ctx->end);
 205
 206         xbt_os_thread_exit(&exit_code);
 207 }
 208
 209 void schedule(ctx_t c)
 210 {
 211         xbt_os_sem_release(c->begin);   /* allow C to go                */
 212         xbt_os_sem_acquire(c->end);     /* wait C's end                 */
 213 }
 214
 215 void unschedule(ctx_t c)
 216 {
 217    xbt_os_sem_release(c->end);          /* I'm done, dude               */
 218    xbt_os_sem_acquire(c->begin);        /* can I start again?   */
 219 }
 220
 221 ctx_t
 222 ctx_new(job_t job)
 223 {
 224         ctx_t ctx = xbt_new0(s_ctx_t,1);
 225         ctx->index = ++__next_ctx_ID;
 226         ctx->begin = xbt_os_sem_init(0);
 227         ctx->end = xbt_os_sem_init(0);
 228         ctx->failure = 0;
 229         ctx->job = job;
 230
 231         return ctx;
 232 }
 233
 234 int
 235 ctx_free(ctx_t* ref)
 236 {
 237         ctx_t ctx;
 238         if(!(*ref))
 239                 return EINVAL;
 240
 241         ctx = *ref;
 242
 243         xbt_os_sem_destroy(ctx->begin);
 244         xbt_os_sem_destroy(ctx->end);
 245         job_free(&(ctx->job));
 246         free(ctx);
 247         *ref = NULL;
 248
 249         return 0;
 250 }
 251
 252 sched_t
 253 sched_new(int size)
 254 {
 255         sched_t sched;
 256
 257         if(size <= 0)
 258         {
 259                 errno = EINVAL;
 260                 return NULL;
 261         }
 262
 263         sched = xbt_new0(s_sched_t,1);
 264
 265         if(!sched)
 266         {
 267                 errno = ENOMEM;
 268                 return NULL;
 269         }
 270
 271         sched->ctxs = xbt_new0(ctx_t,size);
 272
 273         if(!(sched->ctxs))
 274         {
 275                 errno = ENOMEM;
 276                 free(sched);
 277                 return NULL;
 278         }
 279
 280         sched->size = 0;
 281         sched->capacity = size;
 282
 283         return sched;
 284 }
 285
 286 int
 287 sched_add_job(sched_t sched, job_t job)
 288 {
 289         if(!sched || !job)
 290                 return EINVAL;
 291
 292         if(sched->capacity < sched->size)
 293                 return ENOBUFS;
 294
 295         sched->ctxs[(sched->size)++] = ctx_new(job);
 296
 297         return 0;
 298 }
 299
 300 int
 301 sched_init(sched_t sched)
 302 {
 303         int i,j;
 304         int success = 1;
 305
 306         if(!sched)
 307                 return EINVAL;
 308
 309         for(i = 0; i < sched->size; i++)
 310         {
 311                 sched->ctxs[i]->imp = xbt_os_thread_create("thread",ctx_function,(void*)sched->ctxs[i]);
 312
 313                 xbt_os_sem_acquire(sched->ctxs[i]->end);
 314         }
 315
 316         if(!success)
 317         {
 318                 for(j = 0; j < i; j++)
 319                 {
 320                         sched->ctxs[j]->failure = 1;
 321                         xbt_os_sem_release(sched->ctxs[j]->begin);
 322                 }
 323
 324                 for(j = 0; j < i; j++)
 325                 {
 326                         xbt_os_thread_join(sched->ctxs[j]->imp,0);
 327
 328                         ctx_free(&(sched->ctxs[j]));
 329                 }
 330
 331                 return -1;
 332
 333         }
 334
 335         return 0;
 336 }
 337
 338 int
 339 sched_schedule(sched_t sched)
 340 {
 341         int i;
 342
 343         if(!sched)
 344                 return EINVAL;
 345
 346         for(i = 0; i < sched->size; i++)
 347                 schedule(sched->ctxs[i]);
 348
 349         return 0;
 350 }
 351
 352 int
 353 sched_clean(sched_t sched)
 354 {
 355         int i;
 356
 357         if(!sched)
 358                 return EINVAL;
 359
 360         for(i = 0; i < sched->size; i++)
 361         {
 362                 xbt_os_thread_join(sched->ctxs[i]->imp,NULL);
 363
 364                 ctx_free(&(sched->ctxs[i]));
 365         }
 366
 367         return 0;
 368 }
 369
 370 int
 371 sched_free(sched_t* ref)
 372 {
 373         if(*ref)
 374                 return EINVAL;
 375
 376         free(((sched_t)(*ref))->ctxs);
 377
 378         *ref = NULL;
 379
 380         return 0;
 381 }
 382
 383
 384 int
 385 job(int argc, char** argv)
 386 {
 387         int i = 0;
 388
 389         INFO0("I'm the job : I'm going to print all the args of my commande line");
 390
 391         INFO1("-- Arguments (%d):",argc);
 392
 393         for(i = 0; i < argc; i++)
 394                 INFO2("  ---- [%i] %s",i,argv[i]);
 395
 396         return 0;
 397 }
 398
 399 job_t
 400 job_new(pfn_func_t func, int argc, char** argv)
 401 {
 402         job_t job;
 403         int i;
 404
 405         /* todo check the parameters */
 406         job = xbt_new0(s_job_t,1);
 407
 408         if(!job)
 409         {
 410                 errno = ENOMEM;
 411                 return NULL;
 412         }
 413
 414         job->argv = xbt_new0(char*,argc);
 415
 416         if(!(job->argv))
 417         {
 418                 free(job);
 419                 errno = ENOMEM;
 420                 return NULL;
 421         }
 422
 423         for(i = 0; i < argc; i++)
 424                 job->argv[i] = strdup(argv[i]);
 425
 426         job->func = func;
 427         job->argc = argc;
 428
 429         return job;
 430 }
 431
 432 int
 433 job_execute(job_t job)
 434 {
 435         if(!job)
 436                 return EINVAL;
 437
 438         return (*(job->func))(job->argc, job->argv);
 439 }
 440
 441 int
 442 job_free(job_t* ref)
 443 {
 444         job_t job;
 445         int i;
 446
 447         if(!(*ref))
 448                 return EINVAL;
 449
 450         job = *ref;
 451
 452         for(i = 0; i < job->argc; i++)
 453                 free(job->argv[i]);
 454
 455         free(job->argv);
 456         free(*ref);
 457         *ref = NULL;
 458
 459         return 0;
 460 }
 461
 462
 463