examples/xbt/sem_sched.c

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