examples/xbt/sem_sched.c

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