Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Return result on SIMIX_post_io
[simgrid.git] / src / surf / storage.c
1 /* Copyright (c) 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 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 "xbt/ex.h"
8 #include "xbt/dict.h"
9 #include "portable.h"
10 #include "surf_private.h"
11 #include "storage_private.h"
12 #include "surf/surf_resource.h"
13
14 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(surf_storage, surf,
15                                 "Logging specific to the SURF storage module");
16
17 xbt_lib_t storage_lib;
18 int ROUTING_STORAGE_LEVEL;      //Routing for storagelevel
19 int ROUTING_STORAGE_HOST_LEVEL;
20 int SURF_STORAGE_LEVEL;
21 xbt_lib_t storage_type_lib;
22 int ROUTING_STORAGE_TYPE_LEVEL; //Routing for storage_type level
23
24 xbt_dynar_t mount_list = NULL;  /* temporary store of current mount storage */
25
26 surf_model_t surf_storage_model = NULL;
27 lmm_system_t storage_maxmin_system = NULL;
28 static int storage_selective_update = 0;
29 static xbt_swag_t
30     storage_running_action_set_that_does_not_need_being_checked = NULL;
31
32 #define GENERIC_LMM_ACTION(action) action->generic_lmm_action
33 #define GENERIC_ACTION(action) GENERIC_LMM_ACTION(action).generic_action
34
35 typedef enum {
36   READ = 0,
37   WRITE = 1,
38   DEFAULT = 2
39 } STORAGE_execute_type_t;
40
41 typedef struct surf_storage {
42   s_surf_resource_t generic_resource;
43   const char* type;
44   const char* content; /*should be a dict */
45 } s_surf_storage_t, *surf_storage_t;
46
47 static void storage_action_state_set(surf_action_t action, e_surf_action_state_t state);
48 static surf_action_t storage_action_sleep (void *storage, double duration);
49 static surf_action_t storage_action_execute (void *storage, size_t size, STORAGE_execute_type_t type);
50
51 static surf_action_t storage_action_open(void *storage, const char* path, const char* mode)
52 {
53   XBT_DEBUG("\tOpen file '%s'",path);
54   char *storage_type_id = xbt_lib_get_or_null(
55       storage_lib,
56       ((storage_t)storage)->generic_resource.name,
57       ROUTING_STORAGE_LEVEL);
58   storage_type_t storage_type = xbt_lib_get_or_null(storage_type_lib, storage_type_id,ROUTING_STORAGE_TYPE_LEVEL);
59   xbt_dict_t content_dict = storage_type->content;
60   content_t content = xbt_dict_get(content_dict,path);
61
62   surf_file_t file = xbt_new0(s_surf_file_t,1);
63   file->name = xbt_strdup(path);
64   file->content = content;
65
66   surf_action_t action = storage_action_execute(storage,0, DEFAULT);
67   action->file = (void *)file;
68   return action;
69 }
70
71 static surf_action_t storage_action_close(void *storage, surf_file_t fp)
72 {
73   char *filename = fp->name;
74   free(fp->name);
75   fp->content = NULL;
76   xbt_free(fp);
77   surf_action_t action = storage_action_execute(storage,0, DEFAULT);
78   XBT_DEBUG("\tClose file '%s'",filename);
79   return action;
80 }
81
82 static surf_action_t storage_action_read(void *storage, void* ptr, size_t size, size_t nmemb, surf_file_t stream)
83 {
84   char *filename = stream->name;
85   content_t content = stream->content;
86   XBT_DEBUG("\tRead file '%s' size '%Zu/%Zu'",filename,size,content->size);
87   if(size > content->size)
88     size = content->size;
89   surf_action_t action = storage_action_execute(storage,size,READ);
90   return action;
91 }
92
93 static surf_action_t storage_action_write(void *storage, const void* ptr, size_t size, size_t nmemb, surf_file_t stream)
94 {
95   char *filename = stream->name;
96   content_t content = stream->content;
97   XBT_DEBUG("\tWrite file '%s' size '%Zu/%Zu'",filename,size,content->size);
98
99   surf_action_t action = storage_action_execute(storage,size,WRITE);
100   content->size += size;
101   return action;
102 }
103
104 static surf_action_t storage_action_stat(void *storage, int fd, void* buf)
105 {
106   THROW_UNIMPLEMENTED;
107   return NULL;
108 }
109
110 static surf_action_t storage_action_execute (void *storage, size_t size, STORAGE_execute_type_t type)
111 {
112   surf_action_storage_t action = NULL;
113   storage_t STORAGE = storage;
114
115   XBT_IN("(%s,%Zu)", surf_resource_name(STORAGE), size);
116   action =
117       surf_action_new(sizeof(s_surf_action_storage_t), size, surf_storage_model,
118           STORAGE->state_current != SURF_RESOURCE_ON);
119
120   GENERIC_LMM_ACTION(action).suspended = 0;     /* Should be useless because of the
121                                                    calloc but it seems to help valgrind... */
122
123   GENERIC_LMM_ACTION(action).variable =
124       lmm_variable_new(storage_maxmin_system, action, 1.0, -1.0 , 3);
125
126   if(type == DEFAULT)
127   lmm_expand(storage_maxmin_system, STORAGE->constraint,
128              GENERIC_LMM_ACTION(action).variable, 1.0);
129   else if(type == READ)
130     lmm_expand(storage_maxmin_system, STORAGE->constraint_read,
131                GENERIC_LMM_ACTION(action).variable, 1.0);
132   else if(type == WRITE)
133     lmm_expand(storage_maxmin_system, STORAGE->constraint_write,
134                GENERIC_LMM_ACTION(action).variable, 1.0);
135   else xbt_die("storage_action_execute with type '%d'",(int) type);
136
137   XBT_OUT();
138   return (surf_action_t) action;
139 }
140
141 static surf_action_t storage_action_sleep (void *storage, double duration)
142 {
143   surf_action_storage_t action = NULL;
144
145   if (duration > 0)
146     duration = MAX(duration, MAXMIN_PRECISION);
147
148   XBT_IN("(%s,%g)", surf_resource_name(storage), duration);
149   action = (surf_action_storage_t) storage_action_execute(storage, 1.0, DEFAULT);
150   GENERIC_ACTION(action).max_duration = duration;
151   GENERIC_LMM_ACTION(action).suspended = 2;
152   if (duration == NO_MAX_DURATION) {
153     /* Move to the *end* of the corresponding action set. This convention
154        is used to speed up update_resource_state  */
155     xbt_swag_remove(action, ((surf_action_t) action)->state_set);
156     ((surf_action_t) action)->state_set =
157         storage_running_action_set_that_does_not_need_being_checked;
158     xbt_swag_insert(action, ((surf_action_t) action)->state_set);
159   }
160
161   lmm_update_variable_weight(storage_maxmin_system,
162                              GENERIC_LMM_ACTION(action).variable, 0.0);
163
164   XBT_OUT();
165   return (surf_action_t) action;
166 }
167
168 static void* storage_create_resource(const char* id, const char* model,const char* type_id)
169 {
170   storage_t storage = NULL;
171
172   xbt_assert(!surf_storage_resource_by_name(id),
173               "Storage '%s' declared several times in the platform file",
174               id);
175   storage = (storage_t) surf_resource_new(sizeof(s_storage_t),
176           surf_storage_model, id,NULL);
177
178   storage->state_current = SURF_RESOURCE_ON;
179
180   storage_type_t storage_type = xbt_lib_get_or_null(storage_type_lib, type_id,ROUTING_STORAGE_TYPE_LEVEL);
181   size_t Bread  = atof(xbt_dict_get(storage_type->properties,"Bread"));
182   size_t Bwrite = atof(xbt_dict_get(storage_type->properties,"Bwrite"));
183   size_t Bconnexion   = atof(xbt_dict_get(storage_type->properties,"Bconnexion"));
184   XBT_INFO("Create resource with Bconnexion '%Zu' Bread '%Zu' Bwrite '%Zu'",Bconnexion,Bread,Bwrite);
185   storage->constraint       = lmm_constraint_new(storage_maxmin_system, storage, Bconnexion);
186   storage->constraint_read  = lmm_constraint_new(storage_maxmin_system, storage, Bread);
187   storage->constraint_write = lmm_constraint_new(storage_maxmin_system, storage, Bwrite);
188
189   xbt_lib_set(storage_lib, id, SURF_STORAGE_LEVEL, storage);
190
191   XBT_DEBUG("SURF storage create resource\n\t\tid '%s'\n\t\ttype '%s' \n\t\tmodel '%s' \n\t\tproperties '%p'\n\t\tBread '%Zu'\n",
192       id,
193       model,
194       type_id,
195       storage_type->properties,
196       Bread);
197
198   return storage;
199 }
200
201 static void storage_finalize(void)
202 {
203   lmm_system_free(storage_maxmin_system);
204   storage_maxmin_system = NULL;
205
206   surf_model_exit(surf_storage_model);
207   surf_storage_model = NULL;
208
209   xbt_swag_free
210       (storage_running_action_set_that_does_not_need_being_checked);
211   storage_running_action_set_that_does_not_need_being_checked = NULL;
212 }
213
214 static void storage_update_actions_state(double now, double delta)
215 {
216   surf_action_storage_t action = NULL;
217   surf_action_storage_t next_action = NULL;
218   xbt_swag_t running_actions = surf_storage_model->states.running_action_set;
219   xbt_swag_foreach_safe(action, next_action, running_actions) {
220
221     double_update(&(GENERIC_ACTION(action).remains),
222                   lmm_variable_getvalue(GENERIC_LMM_ACTION(action).variable) * delta);
223     if (GENERIC_LMM_ACTION(action).generic_action.max_duration != NO_MAX_DURATION)
224       double_update(&(GENERIC_ACTION(action).max_duration), delta);
225     if ((GENERIC_ACTION(action).remains <= 0) &&
226         (lmm_get_variable_weight(GENERIC_LMM_ACTION(action).variable) > 0)) {
227       GENERIC_ACTION(action).finish = surf_get_clock();
228       storage_action_state_set((surf_action_t) action, SURF_ACTION_DONE);
229     } else if ((GENERIC_ACTION(action).max_duration != NO_MAX_DURATION) &&
230                (GENERIC_ACTION(action).max_duration <= 0)) {
231       GENERIC_ACTION(action).finish = surf_get_clock();
232       storage_action_state_set((surf_action_t) action, SURF_ACTION_DONE);
233     }
234   }
235
236   return;
237 }
238
239 static double storage_share_resources(double NOW)
240 {
241   XBT_DEBUG("storage_share_resources %f",NOW);
242   s_surf_action_storage_t action;
243   return generic_maxmin_share_resources(surf_storage_model->states.running_action_set,
244       xbt_swag_offset(action, generic_lmm_action.variable),
245       storage_maxmin_system, lmm_solve);
246 }
247
248 static int storage_resource_used(void *resource_id)
249 {
250   THROW_UNIMPLEMENTED;
251   return 0;
252 }
253
254 static void storage_resources_state(void *id, tmgr_trace_event_t event_type,
255                                  double value, double time)
256 {
257   THROW_UNIMPLEMENTED;
258 }
259
260 static int storage_action_unref(surf_action_t action)
261 {
262   action->refcount--;
263   if (!action->refcount) {
264     xbt_swag_remove(action, action->state_set);
265     if (((surf_action_lmm_t) action)->variable)
266       lmm_variable_free(storage_maxmin_system,
267                         ((surf_action_lmm_t) action)->variable);
268 #ifdef HAVE_TRACING
269     xbt_free(action->category);
270 #endif
271     surf_action_free(&action);
272     return 1;
273   }
274   return 0;
275 }
276
277 static void storage_action_cancel(surf_action_t action)
278 {
279   surf_action_state_set(action, SURF_ACTION_FAILED);
280   return;
281 }
282
283 static void storage_action_state_set(surf_action_t action, e_surf_action_state_t state)
284 {
285   surf_action_state_set(action, state);
286   return;
287 }
288
289 static void storage_action_suspend(surf_action_t action)
290 {
291   XBT_IN("(%p)", action);
292   if (((surf_action_lmm_t) action)->suspended != 2) {
293     lmm_update_variable_weight(storage_maxmin_system,
294                                ((surf_action_lmm_t) action)->variable,
295                                0.0);
296     ((surf_action_lmm_t) action)->suspended = 1;
297   }
298   XBT_OUT();
299 }
300
301 static void storage_action_resume(surf_action_t action)
302 {
303   THROW_UNIMPLEMENTED;
304 }
305
306 static int storage_action_is_suspended(surf_action_t action)
307 {
308   return (((surf_action_lmm_t) action)->suspended == 1);
309 }
310
311 static void storage_action_set_max_duration(surf_action_t action, double duration)
312 {
313   THROW_UNIMPLEMENTED;
314 }
315
316 static void storage_action_set_priority(surf_action_t action, double priority)
317 {
318   THROW_UNIMPLEMENTED;
319 }
320
321 static void parse_storage_init(sg_platf_storage_cbarg_t storage)
322 {
323   void* stype = xbt_lib_get_or_null(storage_type_lib,
324                                     storage->type_id,
325                                     ROUTING_STORAGE_TYPE_LEVEL);
326   if(!stype) xbt_die("No storage type '%s'",storage->type_id);
327 //  XBT_INFO("SURF storage create resource\n\t\tid '%s'\n\t\ttype '%s' \n\t\tmodel '%s' \n\t\tcontent '%s'\n\t\tproperties '%p'\n",
328 //      storage->id,
329 //      ((storage_type_t) stype)->model,
330 //      ((storage_type_t) stype)->type_id,
331 //      ((storage_type_t) stype)->content,
332 //      ((storage_type_t) stype)->properties);
333
334  storage_create_resource(storage->id,
335      ((storage_type_t) stype)->model,
336      ((storage_type_t) stype)->type_id);
337 }
338
339 static void parse_mstorage_init(sg_platf_mstorage_cbarg_t mstorage)
340 {
341   XBT_DEBUG("parse_mstorage_init");
342 }
343
344 static void parse_storage_type_init(sg_platf_storage_type_cbarg_t storagetype_)
345 {
346   XBT_DEBUG("parse_storage_type_init");
347 }
348
349 static void parse_mount_init(sg_platf_mount_cbarg_t mount)
350 {
351   XBT_DEBUG("parse_mount_init");
352 }
353
354 static void storage_define_callbacks()
355 {
356   sg_platf_storage_add_cb(parse_storage_init);
357   sg_platf_storage_type_add_cb(parse_storage_type_init);
358   sg_platf_mstorage_add_cb(parse_mstorage_init);
359   sg_platf_mount_add_cb(parse_mount_init);
360 }
361
362 static void surf_storage_model_init_internal(void)
363 {
364   s_surf_action_t action;
365
366   XBT_DEBUG("surf_storage_model_init_internal");
367   surf_storage_model = surf_model_init();
368
369   storage_running_action_set_that_does_not_need_being_checked =
370       xbt_swag_new(xbt_swag_offset(action, state_hookup));
371
372   surf_storage_model->name = "Storage";
373   surf_storage_model->action_unref = storage_action_unref;
374   surf_storage_model->action_cancel = storage_action_cancel;
375   surf_storage_model->action_state_set = storage_action_state_set;
376
377   surf_storage_model->model_private->finalize = storage_finalize;
378   surf_storage_model->model_private->update_actions_state = storage_update_actions_state;
379   surf_storage_model->model_private->share_resources = storage_share_resources;
380   surf_storage_model->model_private->resource_used = storage_resource_used;
381   surf_storage_model->model_private->update_resource_state = storage_resources_state;
382
383   surf_storage_model->suspend = storage_action_suspend;
384   surf_storage_model->resume = storage_action_resume;
385   surf_storage_model->is_suspended = storage_action_is_suspended;
386   surf_storage_model->set_max_duration = storage_action_set_max_duration;
387   surf_storage_model->set_priority = storage_action_set_priority;
388
389   surf_storage_model->extension.storage.open = storage_action_open;
390   surf_storage_model->extension.storage.close = storage_action_close;
391   surf_storage_model->extension.storage.read = storage_action_read;
392   surf_storage_model->extension.storage.write = storage_action_write;
393   surf_storage_model->extension.storage.stat = storage_action_stat;
394   surf_storage_model->extension.storage.create_resource = storage_create_resource;
395
396   if (!storage_maxmin_system) {
397     storage_maxmin_system = lmm_system_new(storage_selective_update);
398   }
399
400 }
401
402 void surf_storage_model_init_default(void)
403 {
404   surf_storage_model_init_internal();
405   storage_define_callbacks();
406
407   xbt_dynar_push(model_list, &surf_storage_model);
408 }
409
410 static void storage_parse_storage(sg_platf_storage_cbarg_t storage)
411 {
412   xbt_assert(!xbt_lib_get_or_null(storage_lib, storage->id,ROUTING_STORAGE_LEVEL),
413                "Reading a storage, processing unit \"%s\" already exists", storage->id);
414
415   // Verification of an existing type_id
416 #ifndef NDEBUG
417   void* storage_type = xbt_lib_get_or_null(storage_type_lib, storage->type_id,ROUTING_STORAGE_TYPE_LEVEL);
418 #endif
419   xbt_assert(storage_type,"Reading a storage, type id \"%s\" does not exists", storage->type_id);
420
421   XBT_DEBUG("ROUTING Create a storage name '%s' with type_id '%s'",
422       storage->id,
423       storage->type_id);
424
425   xbt_lib_set(storage_lib,
426       storage->id,
427       ROUTING_STORAGE_LEVEL,
428       (void *) xbt_strdup(storage->type_id));
429 }
430
431 static void free_storage_content(void *p)
432 {
433   content_t content = p;
434   free(content->date);
435   free(content->group);
436   free(content->time);
437   free(content->user);
438   free(content->user_rights);
439   free(content);
440 }
441
442 static xbt_dict_t parse_storage_content(const char *filename)
443 {
444   if ((!filename) || (strcmp(filename, "") == 0))
445     return NULL;
446
447   xbt_dict_t parse_content = xbt_dict_new_homogeneous(free_storage_content);
448   FILE *file = NULL;
449
450   file = surf_fopen(filename, "r");
451   xbt_assert(file != NULL, "Cannot open file '%s' (path=%s)", filename,
452               xbt_str_join(surf_path, ":"));
453
454   char *line = NULL;
455   size_t len = 0;
456   ssize_t read;
457   char user_rights[12];
458   char user[100];
459   char group[100];
460   char date[12];
461   char time[12];
462   char path[1024];
463   int nb, size;
464
465   content_t content;
466
467   while ((read = getline(&line, &len, file)) != -1) {
468     content = xbt_new0(s_content_t,1);
469     if(sscanf(line,"%s %d %s %s %d %s %s %s",user_rights,&nb,user,group,&size,date,time,path)==8) {
470       content->date = xbt_strdup(date);
471       content->group = xbt_strdup(group);
472       content->size = size;
473       content->time = xbt_strdup(time);
474       content->user = xbt_strdup(user);
475       content->user_rights = xbt_strdup(user_rights);
476       xbt_dict_set(parse_content,path,content,NULL);
477     } else {
478       xbt_die("Be sure of passing a good format for content file.\n");
479       // You can generate this kind of file with command line:
480       // find /path/you/want -type f -exec ls -l {} \; 2>/dev/null > ./content.txt
481     }
482   }
483   if (line)
484       free(line);
485
486   fclose(file);
487   return parse_content;
488 }
489
490 static void storage_parse_storage_type(sg_platf_storage_type_cbarg_t storage_type)
491 {
492   xbt_assert(!xbt_lib_get_or_null(storage_type_lib, storage_type->id,ROUTING_STORAGE_TYPE_LEVEL),
493                "Reading a storage type, processing unit \"%s\" already exists", storage_type->id);
494
495   storage_type_t stype = xbt_new0(s_storage_type_t, 1);
496   stype->model = xbt_strdup(storage_type->model);
497   stype->properties = storage_type->properties;
498   stype->content = parse_storage_content(storage_type->content);
499   stype->type_id = xbt_strdup(storage_type->id);
500
501   XBT_DEBUG("ROUTING Create a storage type id '%s' with model '%s' content '%s' and properties '%p'",
502       stype->type_id,
503       stype->model,
504       storage_type->content,
505       stype->properties);
506
507   xbt_lib_set(storage_type_lib,
508       stype->type_id,
509       ROUTING_STORAGE_TYPE_LEVEL,
510       (void *) stype);
511 }
512 static void storage_parse_mstorage(sg_platf_mstorage_cbarg_t mstorage)
513 {
514   THROW_UNIMPLEMENTED;
515 //  mount_t mnt = xbt_new0(s_mount_t, 1);
516 //  mnt->id = xbt_strdup(mstorage->type_id);
517 //  mnt->name = xbt_strdup(mstorage->name);
518 //
519 //  if(!mount_list){
520 //    XBT_DEBUG("Creata a Mount list for %s",A_surfxml_host_id);
521 //    mount_list = xbt_dynar_new(sizeof(char *), NULL);
522 //  }
523 //  xbt_dynar_push(mount_list,(void *) mnt);
524 //  free(mnt->id);
525 //  free(mnt->name);
526 //  xbt_free(mnt);
527 //  XBT_DEBUG("ROUTING Mount a storage name '%s' with type_id '%s'",mstorage->name, mstorage->id);
528 }
529
530 static void mount_free(void *p)
531 {
532   mount_t mnt = p;
533   xbt_free(mnt->name);
534 }
535
536 static void storage_parse_mount(sg_platf_mount_cbarg_t mount)
537 {
538   // Verification of an existing storage
539 #ifndef NDEBUG
540   void* storage = xbt_lib_get_or_null(storage_lib, mount->id,ROUTING_STORAGE_LEVEL);
541 #endif
542   xbt_assert(storage,"Disk id \"%s\" does not exists", mount->id);
543
544   XBT_DEBUG("ROUTING Mount '%s' on '%s'",mount->id, mount->name);
545
546   s_mount_t mnt;
547   mnt.id = surf_storage_resource_by_name(mount->id);
548   mnt.name = xbt_strdup(mount->name);
549
550   if(!mount_list){
551     XBT_DEBUG("Create a Mount list for %s",A_surfxml_host_id);
552     mount_list = xbt_dynar_new(sizeof(s_mount_t), mount_free);
553   }
554   xbt_dynar_push(mount_list,&mnt);
555 }
556
557 static XBT_INLINE void routing_storage_type_free(void *r)
558 {
559   storage_type_t stype = r;
560   free(stype->model);
561   free(stype->type_id);
562   xbt_dict_free(&(stype->properties));
563   xbt_dict_free(&(stype->content));
564   free(stype);
565 }
566
567 static XBT_INLINE void routing_storage_host_free(void *r)
568 {
569   xbt_dynar_t dyn = r;
570   xbt_dynar_free(&dyn);
571 }
572
573 void storage_register_callbacks() {
574
575   ROUTING_STORAGE_LEVEL = xbt_lib_add_level(storage_lib,xbt_free);
576   ROUTING_STORAGE_HOST_LEVEL = xbt_lib_add_level(storage_lib,routing_storage_host_free);
577   ROUTING_STORAGE_TYPE_LEVEL = xbt_lib_add_level(storage_type_lib,routing_storage_type_free);
578   SURF_STORAGE_LEVEL = xbt_lib_add_level(storage_lib,surf_resource_free);
579
580   sg_platf_storage_add_cb(storage_parse_storage);
581   sg_platf_mstorage_add_cb(storage_parse_mstorage);
582   sg_platf_storage_type_add_cb(storage_parse_storage_type);
583   sg_platf_mount_add_cb(storage_parse_mount);
584 }