frsh_api/frsh_thread.c

   1 /**
   2  * @file   frsh_thread.c
   3  * @author Dario Faggioli <faggioli@gandalf.sssup.it>
   4  *
   5  * @brief  FRSH core thread related functions not implamented in managers..
   6  *
   7  *
   8  */
   9 #include <semaphore.h>
  10
  11 #include <fres_contract.h>
  12 #include <fres_contract_idl.h>
  13 #include <frsh_core.h>
  14 #include <frsh_error.h>
  15 #include <fres_blocks.h>
  16 #include <string.h>
  17 #include <fcb.h>
  18 #include <fra_generic.h>
  19
  20 #include "frsh_forb.h"
  21
  22 #ifdef CONFIG_AQUOSA
  23 #include <aqcpu_res.h>
  24 #endif
  25 #ifdef CONFIG_CPUCG
  26 #include <cpucg_res.h>
  27 #endif
  28 #ifdef CONFIG_DISKBFQ
  29 #include <diskbfq_res.h>
  30 #endif
  31
  32 /**
  33  * frsh_thread_bind(), bind a thread to a vres
  34  *
  35  * possible return values:
  36  *  FRSH_NO_ERROR
  37  *  FRSH_ERR_NOT_INITIALIZED
  38  *  FRSH_BAD_ARGUMENT(*) (invalid or dummy vres)
  39  *  FRSH_ERR_NOT_CONTRACTED_VRES(*)
  40  *  FRSH_ERR_ALREADY_BOUND(*)
  41  *  FRSH_ERR_INTERNAL_ERROR(*) (something wrong with AQuoSA or with the service thread internal data structures)
  42  */
  43 int frsh_thread_bind
  44   (const frsh_vres_id_t vres,
  45    const frsh_thread_id_t thread)
  46 {
  47         int ret;
  48         ret = fra_insert_thread_vres(&thread, vres->allocator->res_type, vres);
  49         if (ret) goto err;
  50
  51         if (vres->allocator->bind_thread) {
  52                 ret = vres->allocator->bind_thread(vres, thread);
  53         } else {
  54                 ret = FRSH_ERR_NOT_IMPLEMENTED;
  55         }
  56         if (ret) goto err_delete;
  57
  58         return FRSH_NO_ERROR;
  59
  60 err_delete:
  61         fra_delete_thread_vres(&thread, vres->allocator->res_type);
  62 err:
  63         return FRSH_ERR_INTERNAL_ERROR;
  64 }
  65
  66 /*
  67  * frsh_thread_unbind(), unbind a thread from a vres
  68  *
  69  * @FIXME:
  70  *  Notice that, since we don't know from which VRES the
  71  *   thread should be unbound, we unboind it from all the
  72  *   ones it is attached to!
  73  *
  74  * possible return values:
  75  *  FRSH_NO_ERROR
  76  *  FRSH_ERR_NOT_INITIALIZED
  77  *  FRSH_BAD_ARGUMENT(*) (invalid thread)
  78  *  FRSH_ERR_NOT_BOUND(*)
  79  *  FRSH_ERR_INTERNAL_ERROR(*) (something wrong with AQuoSA or with the service thread internal data structures)
  80  */
  81 int frsh_thread_unbind(const frsh_thread_id_t thread)
  82 {
  83         fres_vres_t *vres;
  84         int ret;
  85
  86         /* Unbound from FRSH_RT_PROCESSOR resource (if any) */
  87         vres = fra_get_vres_thread_vres(&thread, FRSH_RT_PROCESSOR);
  88         if (vres) {
  89                 if (vres->allocator->unbind_thread) {
  90                         ret = vres->allocator->unbind_thread(thread);
  91                 } else {
  92                         ret = FRSH_ERR_NOT_IMPLEMENTED;
  93                 }
  94                 if (ret) goto err;
  95
  96                 ret = fra_delete_thread_vres(&thread, vres->allocator->res_type);
  97                 if (ret) goto err;
  98         }
  99
 100         /* Unbound from FRSH_RT_DISK resource (if any) */
 101         vres = fra_get_vres_thread_vres(&thread, FRSH_RT_DISK);
 102         if (vres) {
 103                 if (vres->allocator->unbind_thread) {
 104                         ret = vres->allocator->unbind_thread(thread);
 105                 } else {
 106                         ret = FRSH_ERR_NOT_IMPLEMENTED;
 107                 }
 108                 if (ret) goto err;
 109
 110                 ret = fra_delete_thread_vres(&thread, vres->allocator->res_type);
 111                 if (ret) goto err;
 112         }
 113
 114         return FRSH_NO_ERROR;
 115 err:
 116         return FRSH_ERR_INTERNAL_ERROR;
 117 }
 118
 119 /**
 120  * structure used to pass the new created thread (or better its wrapper) the
 121  * arguments it needs to bind itself to the vres.
 122  */
 123 typedef struct wrapper_pthread_arg {
 124         frsh_thread_id_t parent_thread_id;
 125         frsh_thread_id_t *thread_id;
 126         frsh_thread_code_t pthread_code;        /**< function to be running in thread */
 127         void *pthread_arg;                      /**< thread arguments*/
 128         frsh_vres_id_t vres;
 129         sem_t *stopper;                         /**< semaphore to synchronize frsh call and wrapper thread */
 130         int  errcode;
 131 } wrapper_pthread_arg_t;
 132
 133 /*
 134  * typedef struct {
 135  *      pthread_t pthread_id;
 136  *      pid_t linux_pid;
 137  *      pid_t linux_tid;
 138  * } FOSA_THREAD_ID_T_OPAQUE;
 139  */
 140
 141 /**
 142  * Wrapper thread
 143  *
 144  * possible exit status:
 145  *  FRSH_ERR_INTENRAL_ERROR (something wrong with threads and signal handling)
 146  *  FRSH_ERR_TOO_MANY_TASKS(*)
 147  *  FRSH_ERR_BAD_ARGUMENT(*) (invalid vres)
 148  *  FRSH_ERR_NOT_CONTRACTED_VRES(*)
 149  *  whatever the user provided code 'pthread_code' returns
 150  */
 151 static void* wrapper_pthread(void *arg)
 152 {
 153         wrapper_pthread_arg_t* pth = (wrapper_pthread_arg_t*) arg;
 154         frsh_thread_id_t* thread_id = pth->thread_id;
 155         void *thread_arg = pth->pthread_arg;
 156         frsh_thread_code_t thread_code = pth->pthread_code;
 157
 158         /* bind this thread to vres */
 159         *thread_id = fosa_thread_self();
 160         pth->errcode = frsh_thread_bind(pth->vres, *thread_id);
 161
 162         sem_post(pth->stopper);
 163         /* No access to pth is allowed after this point, it can vanish already */
 164
 165         if (pth->errcode) return NULL;
 166
 167         /* execute thread code */
 168         return thread_code(thread_arg);
 169 }
 170
 171 /*
 172  * API call for 'frsh_thread_create_and_bind()', as said prepares the
 173  * wrapper code argument data structure, create the new thread and wait
 174  * its acknowledgment before stepping over
 175  *
 176  * possible return values:
 177  *  FRSH_NO_ERROR
 178  *  FRSH_ERR_NOT_INITIALIZED
 179  *  FRSH_ERR_BAD_ARGUMENT (NULL thread or thread_code)
 180  *  whatever 'fosa_thread_create()' returns
 181  *  FRSH_ERR_INTERNAL_ERROR
 182  */
 183 int frsh_thread_create_and_bind
 184   (const frsh_vres_id_t vres,
 185    frsh_thread_id_t *thread,
 186    frsh_thread_attr_t *attr,
 187    frsh_thread_code_t thread_code,
 188    void *arg)
 189 {
 190         wrapper_pthread_arg_t wp_arg;
 191         sem_t stopper;
 192         int ret;
 193
 194         sem_init(&stopper, 0, 0);
 195
 196         if (!thread || !thread_code || !attr)
 197                 return FRSH_ERR_BAD_ARGUMENT;
 198
 199         wp_arg.parent_thread_id = fosa_thread_self();
 200         wp_arg.thread_id = thread;
 201         wp_arg.pthread_code = thread_code;
 202         wp_arg.pthread_arg = arg;
 203         wp_arg.vres = vres;
 204         wp_arg.stopper = &stopper;
 205
 206         /* create the wrapper thread */
 207         pthread_attr_setinheritsched(attr, PTHREAD_EXPLICIT_SCHED);
 208         pthread_attr_setscope(attr, PTHREAD_SCOPE_SYSTEM);
 209         ret = fosa_thread_create(thread, attr, wrapper_pthread,
 210                                 (void*)&wp_arg);
 211
 212         if (ret != 0)
 213                 return(ret);
 214
 215         sem_wait(&stopper);
 216
 217         return (wp_arg.errcode);
 218 }
 219
 220 int frsh_thread_create_in_background
 221   (frsh_thread_code_t thread_code,
 222    const void *thread_arg,
 223    const frsh_contract_label_t contract_label,
 224    frsh_thread_attr_t *attr,
 225    frsh_thread_id_t *thread,
 226    frsh_vres_id_t *vres_id)
 227 {
 228         frsh_contract_t contract;
 229         fosa_rel_time_t zero_msec = fosa_msec_to_rel_time(0);
 230         int ret = 0;
 231
 232         /* Background contract creation and negotiation */
 233         frsh_contract_init(&contract);
 234         frsh_contract_set_resource_and_label(&contract,
 235                                              FRSH_RT_PROCESSOR,
 236                                              0,
 237                                              contract_label);
 238         frsh_contract_set_basic_params(&contract,
 239                                        &zero_msec,
 240                                        &zero_msec,
 241                                        FRSH_WT_INDETERMINATE,
 242                                        FRSH_CT_BACKGROUND);
 243
 244         /* It will be accepted: we are asking for 0 budget over 0 period */
 245         ret = frsh_contract_negotiate(&contract, vres_id);
 246         if (ret !=0) goto error;
 247
 248         ret = fosa_thread_create(thread, attr, thread_code, (void*) thread_arg);
 249         if (ret != 0) goto error;
 250
 251         ret = fra_insert_thread_vres(thread, FRSH_RT_PROCESSOR, *vres_id);
 252         if (ret) goto error;
 253
 254         return 0;
 255 error:
 256         return ret;
 257 }
 258
 259 int frsh_thread_get_vres_id
 260   (const frsh_thread_id_t thread,
 261    frsh_vres_id_t *vres_id)
 262 {
 263         if (!vres_id)
 264                 return FRSH_ERR_BAD_ARGUMENT;
 265
 266         /*
 267          * @FIXME:
 268          *  Why PROCESSOR and not DISK or NETWORK (or whatever) ??
 269          */
 270         *vres_id = fra_get_vres_thread_vres(&thread,
 271                                             FRSH_RT_PROCESSOR);
 272
 273         return vres_id ? 0 : EINVAL;
 274 }
 275
 276 int frsh_service_thread_set_data
 277   (const struct timespec *budget,
 278    const struct timespec *period,
 279    bool *accepted)
 280 {
 281         return FRSH_ERR_NOT_IMPLEMENTED;
 282 }
 283
 284 int frsh_service_thread_get_data
 285   (frsh_rel_time_t *budget,
 286    frsh_rel_time_t *period)
 287 {
 288         return FRSH_ERR_NOT_IMPLEMENTED;
 289 }
 290