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