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                 xbt_exit();
 131                 exit(EXIT_FAILURE);
 132         }
 133
 134         __argv = xbt_new0(char*,MAX_ARGS);
 135
 136         for(i = 0; i < MAX_ARGS; i++)
 137         {
 138                 sprintf(arg,"arg_%d",i);
 139                 __argv[i] = strdup(arg);
 140                 memset(arg,0,MAX_ARG);
 141
 142         }
 143
 144         for(i = 0; i < size; i++)
 145                 sched_add_job(sched,job_new(job,(i < MAX_ARGS) ? i : MAX_ARGS,__argv));
 146
 147         /* initialize the scheduler */
 148         if(sched_init(sched) < 0)
 149         {
 150                 sched_free(&sched);
 151                 INFO1("sched_init() failed : errno %d\n",errno);
 152                 xbt_exit();
 153                 exit(EXIT_FAILURE);
 154         }
 155
 156         /* schedule the jobs */
 157         if(sched_schedule(sched) < 0)
 158         {
 159                 sched_free(&sched);
 160                 INFO1("sched_init() failed : errno %d",errno);
 161                 xbt_exit();
 162                 exit(EXIT_FAILURE);
 163         }
 164
 165         /* cleanup */
 166         if(sched_clean(sched) < 0)
 167         {
 168                 sched_free(&sched);
 169                 INFO1("sched_init() failed : errno %d",errno);
 170                 xbt_exit();
 171                 exit(EXIT_FAILURE);
 172         }
 173
 174         /* destroy the scheduler */
 175         sched_free(&sched);
 176
 177         INFO1("sem_sched terminated with exit code %d (success)",EXIT_SUCCESS);
 178
 179         xbt_exit();
 180
 181         return EXIT_SUCCESS;
 182
 183 }
 184
 185 void*
 186 ctx_function(void* param)
 187 {
 188         int i = 0;
 189         int exit_code = 1;
 190         ctx_t ctx = (ctx_t)param;
 191
 192         INFO1("Hello i'm the owner of the context %d, i'm waiting for starting",ctx->index);
 193
 194         unschedule(ctx);
 195
 196         if(ctx->failure)
 197         {
 198                 INFO1("0ups the scheduler initialization failed bye {%d}.",ctx->index);
 199                 xbt_os_thread_exit(&exit_code);
 200         }
 201
 202         INFO1("I'm the owner of the context %d : I'm started",ctx->index);
 203         INFO0("Wait a minute, I do my job");
 204
 205         /* do its job */
 206         exit_code = job_execute(ctx->job);
 207
 208         INFO1("Have finished my job, bye {%d}\n",ctx->index);
 209
 210         xbt_os_sem_release(ctx->end);
 211
 212         xbt_os_thread_exit(&exit_code);
 213 }
 214
 215 void schedule(ctx_t c)
 216 {
 217         xbt_os_sem_release(c->begin);   /* allow C to go                */
 218         xbt_os_sem_acquire(c->end);     /* wait C's end                 */
 219 }
 220
 221 void unschedule(ctx_t c)
 222 {
 223    xbt_os_sem_release(c->end);          /* I'm done, dude               */
 224    xbt_os_sem_acquire(c->begin);        /* can I start again?   */
 225 }
 226
 227 ctx_t
 228 ctx_new(job_t job)
 229 {
 230         ctx_t ctx = xbt_new0(s_ctx_t,1);
 231         ctx->index = ++__next_ctx_ID;
 232         ctx->begin = xbt_os_sem_init(0);
 233         ctx->end = xbt_os_sem_init(0);
 234         ctx->failure = 0;
 235         ctx->job = job;
 236
 237         return ctx;
 238 }
 239
 240 int
 241 ctx_free(ctx_t* ref)
 242 {
 243         ctx_t ctx;
 244         if(!(*ref))
 245                 return EINVAL;
 246
 247         ctx = *ref;
 248
 249         xbt_os_sem_destroy(ctx->begin);
 250         xbt_os_sem_destroy(ctx->end);
 251         job_free(&(ctx->job));
 252         free(ctx);
 253         *ref = NULL;
 254
 255         return 0;
 256 }
 257
 258 sched_t
 259 sched_new(int size)
 260 {
 261         sched_t sched;
 262
 263         if(size <= 0)
 264         {
 265                 errno = EINVAL;
 266                 return NULL;
 267         }
 268
 269         sched = xbt_new0(s_sched_t,1);
 270
 271         if(!sched)
 272         {
 273                 errno = ENOMEM;
 274                 return NULL;
 275         }
 276
 277         sched->ctxs = xbt_new0(ctx_t,size);
 278
 279         if(!(sched->ctxs))
 280         {
 281                 errno = ENOMEM;
 282                 free(sched);
 283                 return NULL;
 284         }
 285
 286         sched->size = 0;
 287         sched->capacity = size;
 288
 289         return sched;
 290 }
 291
 292 int
 293 sched_add_job(sched_t sched, job_t job)
 294 {
 295         if(!sched || !job)
 296                 return EINVAL;
 297
 298         if(sched->capacity < sched->size)
 299                 return ENOBUFS;
 300
 301         sched->ctxs[(sched->size)++] = ctx_new(job);
 302
 303         return 0;
 304 }
 305
 306 int
 307 sched_init(sched_t sched)
 308 {
 309         int i,j;
 310         int success = 1;
 311
 312         if(!sched)
 313                 return EINVAL;
 314
 315         for(i = 0; i < sched->size; i++)
 316         {
 317                 sched->ctxs[i]->imp = xbt_os_thread_create("thread",ctx_function,(void*)sched->ctxs[i]);
 318
 319                 xbt_os_sem_acquire(sched->ctxs[i]->end);
 320         }
 321
 322         if(!success)
 323         {
 324                 for(j = 0; j < i; j++)
 325                 {
 326                         sched->ctxs[j]->failure = 1;
 327                         xbt_os_sem_release(sched->ctxs[j]->begin);
 328                 }
 329
 330                 for(j = 0; j < i; j++)
 331                 {
 332                         xbt_os_thread_join(sched->ctxs[j]->imp,0);
 333
 334                         ctx_free(&(sched->ctxs[j]));
 335                 }
 336
 337                 return -1;
 338
 339         }
 340
 341         return 0;
 342 }
 343
 344 int
 345 sched_schedule(sched_t sched)
 346 {
 347         int i;
 348
 349         if(!sched)
 350                 return EINVAL;
 351
 352         for(i = 0; i < sched->size; i++)
 353                 schedule(sched->ctxs[i]);
 354
 355         return 0;
 356 }
 357
 358 int
 359 sched_clean(sched_t sched)
 360 {
 361         int i;
 362
 363         if(!sched)
 364                 return EINVAL;
 365
 366         for(i = 0; i < sched->size; i++)
 367         {
 368                 xbt_os_thread_join(sched->ctxs[i]->imp,NULL);
 369
 370                 ctx_free(&(sched->ctxs[i]));
 371         }
 372
 373         return 0;
 374 }
 375
 376 int
 377 sched_free(sched_t* ref)
 378 {
 379         if(*ref)
 380                 return EINVAL;
 381
 382         free(((sched_t)(*ref))->ctxs);
 383
 384         *ref = NULL;
 385
 386         return 0;
 387 }
 388
 389
 390 int
 391 job(int argc, char** argv)
 392 {
 393         int i = 0;
 394
 395         INFO0("I'm the job : I'm going to print all the args of my commande line");
 396
 397         INFO1("-- Arguments (%d):",argc);
 398
 399         for(i = 0; i < argc; i++)
 400                 INFO2("  ---- [%i] %s",i,argv[i]);
 401
 402         return 0;
 403 }
 404
 405 job_t
 406 job_new(pfn_func_t func, int argc, char** argv)
 407 {
 408         job_t job;
 409         int i;
 410
 411         /* todo check the parameters */
 412         job = xbt_new0(s_job_t,1);
 413
 414         if(!job)
 415         {
 416                 errno = ENOMEM;
 417                 return NULL;
 418         }
 419
 420         job->argv = xbt_new0(char*,argc);
 421
 422         if(!(job->argv))
 423         {
 424                 free(job);
 425                 errno = ENOMEM;
 426                 return NULL;
 427         }
 428
 429         for(i = 0; i < argc; i++)
 430                 job->argv[i] = strdup(argv[i]);
 431
 432         job->func = func;
 433         job->argc = argc;
 434
 435         return job;
 436 }
 437
 438 int
 439 job_execute(job_t job)
 440 {
 441         if(!job)
 442                 return EINVAL;
 443
 444         return (*(job->func))(job->argc, job->argv);
 445 }
 446
 447 int
 448 job_free(job_t* ref)
 449 {
 450         job_t job;
 451         int i;
 452
 453         if(!(*ref))
 454                 return EINVAL;
 455
 456         job = *ref;
 457
 458         for(i = 0; i < job->argc; i++)
 459                 free(job->argv[i]);
 460
 461         free(job->argv);
 462         free(*ref);
 463         *ref = NULL;
 464
 465         return 0;
 466 }
 467
 468
 469