Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Fix bug and leak for msg_file_open
[simgrid.git] / src / surf / storage_n11.cpp
1 /* Copyright (c) 2013-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 "storage_n11.hpp"
8 #include "surf_private.h"
9
10 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(surf_storage);
11
12 static int storage_selective_update = 0;
13 static xbt_swag_t storage_running_action_set_that_does_not_need_being_checked = NULL;
14
15 /*************
16  * CallBacks *
17  *************/
18
19 static XBT_INLINE void routing_storage_type_free(void *r)
20 {
21   storage_type_t stype = (storage_type_t) r;
22   free(stype->model);
23   free(stype->type_id);
24   free(stype->content);
25   free(stype->content_type);
26   xbt_dict_free(&(stype->properties));
27   xbt_dict_free(&(stype->model_properties));
28   free(stype);
29 }
30
31 static XBT_INLINE void surf_storage_resource_free(void *r)
32 {
33   // specific to storage
34   StoragePtr storage = static_cast<StoragePtr>(r);
35   // generic resource
36   delete storage;
37 }
38
39 static XBT_INLINE void routing_storage_host_free(void *r)
40 {
41   xbt_dynar_t dyn = (xbt_dynar_t) r;
42   xbt_dynar_free(&dyn);
43 }
44
45 static void parse_storage_init(sg_platf_storage_cbarg_t storage)
46 {
47   void* stype = xbt_lib_get_or_null(storage_type_lib,
48                                     storage->type_id,
49                                     ROUTING_STORAGE_TYPE_LEVEL);
50   if(!stype) xbt_die("No storage type '%s'",storage->type_id);
51
52   // if storage content is not specified use the content of storage_type if exist
53   if(!strcmp(storage->content,"") && strcmp(((storage_type_t) stype)->content,"")){
54     storage->content = ((storage_type_t) stype)->content;
55     storage->content_type = ((storage_type_t) stype)->content_type;
56     XBT_DEBUG("For disk '%s' content is empty, inherit the content (of type %s) from storage type '%s' ",
57         storage->id,((storage_type_t) stype)->content_type,
58         ((storage_type_t) stype)->type_id);
59   }
60
61   XBT_DEBUG("SURF storage create resource\n\t\tid '%s'\n\t\ttype '%s' "
62       "\n\t\tmodel '%s' \n\t\tcontent '%s'\n\t\tcontent_type '%s' "
63       "\n\t\tproperties '%p''\n",
64       storage->id,
65       ((storage_type_t) stype)->model,
66       ((storage_type_t) stype)->type_id,
67       storage->content,
68       storage->content_type,
69       storage->properties);
70
71   surf_storage_model->createResource(storage->id,
72                                      ((storage_type_t) stype)->type_id,
73                                      storage->content,
74                                      storage->content_type,
75                                      storage->properties,
76                                      storage->attach);
77 }
78
79 static void parse_mstorage_init(sg_platf_mstorage_cbarg_t /*mstorage*/)
80 {
81   XBT_DEBUG("parse_mstorage_init");
82 }
83
84 static void parse_storage_type_init(sg_platf_storage_type_cbarg_t /*storagetype_*/)
85 {
86   XBT_DEBUG("parse_storage_type_init");
87 }
88
89 static void parse_mount_init(sg_platf_mount_cbarg_t /*mount*/)
90 {
91   XBT_DEBUG("parse_mount_init");
92 }
93
94 static void storage_parse_storage(sg_platf_storage_cbarg_t storage)
95 {
96   xbt_assert(!xbt_lib_get_or_null(storage_lib, storage->id,ROUTING_STORAGE_LEVEL),
97                "Reading a storage, processing unit \"%s\" already exists", storage->id);
98
99   // Verification of an existing type_id
100 #ifndef NDEBUG
101   void* storage_type = xbt_lib_get_or_null(storage_type_lib, storage->type_id,ROUTING_STORAGE_TYPE_LEVEL);
102 #endif
103   xbt_assert(storage_type,"Reading a storage, type id \"%s\" does not exists", storage->type_id);
104
105   XBT_DEBUG("ROUTING Create a storage name '%s' with type_id '%s' and content '%s'",
106       storage->id,
107       storage->type_id,
108       storage->content);
109
110   xbt_lib_set(storage_lib,
111       storage->id,
112       ROUTING_STORAGE_LEVEL,
113       (void *) xbt_strdup(storage->type_id));
114 }
115
116 static void storage_parse_storage_type(sg_platf_storage_type_cbarg_t storage_type)
117 {
118   xbt_assert(!xbt_lib_get_or_null(storage_type_lib, storage_type->id,ROUTING_STORAGE_TYPE_LEVEL),
119                "Reading a storage type, processing unit \"%s\" already exists", storage_type->id);
120
121   storage_type_t stype = xbt_new0(s_storage_type_t, 1);
122   stype->model = xbt_strdup(storage_type->model);
123   stype->properties = storage_type->properties;
124   stype->content = xbt_strdup(storage_type->content);
125   stype->content_type = xbt_strdup(storage_type->content_type);
126   stype->type_id = xbt_strdup(storage_type->id);
127   stype->size = storage_type->size;
128   stype->model_properties = storage_type->model_properties;
129
130   XBT_DEBUG("ROUTING Create a storage type id '%s' with model '%s', "
131       "content '%s', and content_type '%s'",
132       stype->type_id,
133       stype->model,
134       storage_type->content,
135       storage_type->content_type);
136
137   xbt_lib_set(storage_type_lib,
138       stype->type_id,
139       ROUTING_STORAGE_TYPE_LEVEL,
140       (void *) stype);
141 }
142
143 static void storage_parse_mstorage(sg_platf_mstorage_cbarg_t /*mstorage*/)
144 {
145   THROW_UNIMPLEMENTED;
146 //  mount_t mnt = xbt_new0(s_mount_t, 1);
147 //  mnt->id = xbt_strdup(mstorage->type_id);
148 //  mnt->name = xbt_strdup(mstorage->name);
149 //
150 //  if(!mount_list){
151 //    XBT_DEBUG("Creata a Mount list for %s",A_surfxml_host_id);
152 //    mount_list = xbt_dynar_new(sizeof(char *), NULL);
153 //  }
154 //  xbt_dynar_push(mount_list,(void *) mnt);
155 //  free(mnt->id);
156 //  free(mnt->name);
157 //  xbt_free(mnt);
158 //  XBT_DEBUG("ROUTING Mount a storage name '%s' with type_id '%s'",mstorage->name, mstorage->id);
159 }
160
161 static void mount_free(void *p)
162 {
163   mount_t mnt = (mount_t) p;
164   xbt_free(mnt->name);
165 }
166
167 static void storage_parse_mount(sg_platf_mount_cbarg_t mount)
168 {
169   // Verification of an existing storage
170 #ifndef NDEBUG
171   void* storage = xbt_lib_get_or_null(storage_lib, mount->storageId, ROUTING_STORAGE_LEVEL);
172 #endif
173   xbt_assert(storage,"Disk id \"%s\" does not exists", mount->storageId);
174
175   XBT_DEBUG("ROUTING Mount '%s' on '%s'",mount->storageId, mount->name);
176
177   s_mount_t mnt;
178   mnt.storage = surf_storage_resource_priv(surf_storage_resource_by_name(mount->storageId));
179   mnt.name = xbt_strdup(mount->name);
180
181   if(!mount_list){
182     XBT_DEBUG("Create a Mount list for %s",A_surfxml_host_id);
183     mount_list = xbt_dynar_new(sizeof(s_mount_t), mount_free);
184   }
185   xbt_dynar_push(mount_list, &mnt);
186 }
187
188 static void storage_define_callbacks()
189 {
190   sg_platf_storage_add_cb(parse_storage_init);
191   sg_platf_storage_type_add_cb(parse_storage_type_init);
192   sg_platf_mstorage_add_cb(parse_mstorage_init);
193   sg_platf_mount_add_cb(parse_mount_init);
194 }
195
196 void storage_register_callbacks() {
197
198   ROUTING_STORAGE_LEVEL = xbt_lib_add_level(storage_lib,xbt_free);
199   ROUTING_STORAGE_HOST_LEVEL = xbt_lib_add_level(storage_lib, routing_storage_host_free);
200   ROUTING_STORAGE_TYPE_LEVEL = xbt_lib_add_level(storage_type_lib, routing_storage_type_free);
201   SURF_STORAGE_LEVEL = xbt_lib_add_level(storage_lib, surf_storage_resource_free);
202
203   sg_platf_storage_add_cb(storage_parse_storage);
204   sg_platf_mstorage_add_cb(storage_parse_mstorage);
205   sg_platf_storage_type_add_cb(storage_parse_storage_type);
206   sg_platf_mount_add_cb(storage_parse_mount);
207 }
208
209 /*********
210  * Model *
211  *********/
212
213 void surf_storage_model_init_default(void)
214 {
215   surf_storage_model = new StorageN11Model();
216   storage_define_callbacks();
217   xbt_dynar_push(model_list, &surf_storage_model);
218 }
219
220 StorageN11Model::StorageN11Model() : StorageModel() {
221   ActionPtr action = NULL;
222
223   XBT_DEBUG("surf_storage_model_init_internal");
224
225   storage_running_action_set_that_does_not_need_being_checked =
226       xbt_swag_new(xbt_swag_offset(*action, p_stateHookup));
227   if (!p_maxminSystem) {
228     p_maxminSystem = lmm_system_new(storage_selective_update);
229   }
230 }
231
232 StorageN11Model::~StorageN11Model(){
233   xbt_swag_free(storage_running_action_set_that_does_not_need_being_checked);
234   storage_running_action_set_that_does_not_need_being_checked = NULL;
235 }
236
237 StoragePtr StorageN11Model::createResource(const char* id, const char* type_id,
238                 const char* content_name, const char* content_type, xbt_dict_t properties, const char* attach)
239 {
240
241   xbt_assert(!surf_storage_resource_priv(surf_storage_resource_by_name(id)),
242               "Storage '%s' declared several times in the platform file",
243               id);
244
245   storage_type_t storage_type = (storage_type_t) xbt_lib_get_or_null(storage_type_lib, type_id,ROUTING_STORAGE_TYPE_LEVEL);
246
247   double Bread  = surf_parse_get_bandwidth((char*)xbt_dict_get(storage_type->model_properties, "Bread"));
248   double Bwrite = surf_parse_get_bandwidth((char*)xbt_dict_get(storage_type->model_properties, "Bwrite"));
249   double Bconnection   = surf_parse_get_bandwidth((char*)xbt_dict_get(storage_type->model_properties, "Bconnection"));
250
251   StoragePtr storage = new StorageN11(this, id, properties, p_maxminSystem,
252                   Bread, Bwrite, Bconnection,
253                   type_id, (char *)content_name, xbt_strdup(content_type), storage_type->size, (char *) attach);
254
255   xbt_lib_set(storage_lib, id, SURF_STORAGE_LEVEL, static_cast<ResourcePtr>(storage));
256
257   XBT_DEBUG("SURF storage create resource\n\t\tid '%s'\n\t\ttype '%s'\n\t\tproperties '%p'\n\t\tBread '%f'\n",
258       id,
259       type_id,
260       properties,
261       Bread);
262
263   if(!p_storageList)
264         p_storageList = xbt_dynar_new(sizeof(char *),NULL);
265   xbt_dynar_push(p_storageList, &storage);
266
267   return storage;
268 }
269
270 double StorageN11Model::shareResources(double now)
271 {
272   XBT_DEBUG("storage_share_resources %f", now);
273   unsigned int i, j;
274   StoragePtr storage;
275   void *_write_action;
276   StorageActionPtr write_action;
277
278   double min_completion = shareResourcesMaxMin(getRunningActionSet(),
279       p_maxminSystem, lmm_solve);
280
281   double rate;
282   // Foreach disk
283   xbt_dynar_foreach(p_storageList,i,storage)
284   {
285     rate = 0;
286     // Foreach write action on disk
287     xbt_dynar_foreach(storage->p_writeActions, j, _write_action)
288     {
289       write_action = static_cast<StorageActionPtr>(_write_action);
290       rate += lmm_variable_getvalue(write_action->getVariable());
291     }
292     if(rate > 0)
293       min_completion = MIN(min_completion, (storage->m_size-storage->m_usedSize)/rate);
294   }
295
296   return min_completion;
297 }
298
299 void StorageN11Model::updateActionsState(double /*now*/, double delta)
300 {
301   StorageActionPtr action = NULL;
302
303   ActionListPtr actionSet = getRunningActionSet();
304   for(ActionList::iterator it(actionSet->begin()), itNext=it, itend(actionSet->end())
305      ; it != itend ; it=itNext) {
306     ++itNext;
307     action = static_cast<StorageActionPtr>(&*it);
308     if(action->m_type == WRITE)
309     {
310       // Update the disk usage
311      // Update the file size
312      // For each action of type write
313       double rate = lmm_variable_getvalue(action->getVariable());
314       /* Hack to avoid rounding differences between x86 and x86_64
315        * (note that the next sizes are of type sg_size_t). */
316       long incr = delta * rate + MAXMIN_PRECISION;
317       action->p_storage->m_usedSize += incr; // disk usage
318       action->p_file->size += incr; // file size
319
320       sg_size_t *psize = xbt_new(sg_size_t,1);
321       *psize = action->p_file->size;
322       xbt_dict_t content_dict = action->p_storage->p_content;
323       xbt_dict_set(content_dict, action->p_file->name, psize, NULL);
324     }
325
326     action->updateRemains(lmm_variable_getvalue(action->getVariable()) * delta);
327
328     if (action->getMaxDuration() != NO_MAX_DURATION)
329       action->updateMaxDuration(delta);
330
331     if(action->getRemainsNoUpdate() > 0 &&
332         lmm_get_variable_weight(action->getVariable()) > 0 &&
333         action->p_storage->m_usedSize == action->p_storage->m_size)
334     {
335       action->finish();
336       action->setState(SURF_ACTION_FAILED);
337     } else if ((action->getRemainsNoUpdate() <= 0) &&
338         (lmm_get_variable_weight(action->getVariable()) > 0))
339     {
340       action->finish();
341       action->setState(SURF_ACTION_DONE);
342     } else if ((action->getMaxDuration() != NO_MAX_DURATION) &&
343                (action->getMaxDuration() <= 0))
344     {
345       action->finish();
346       action->setState(SURF_ACTION_DONE);
347     }
348   }
349   return;
350 }
351
352 /************
353  * Resource *
354  ************/
355
356 StorageN11::StorageN11(StorageModelPtr model, const char* name, xbt_dict_t properties,
357              lmm_system_t maxminSystem, double bread, double bwrite, double bconnection,
358              const char* type_id, char *content_name, char *content_type, sg_size_t size, char *attach)
359  : Storage(model, name, properties,
360            maxminSystem, bread, bwrite, bconnection, type_id, content_name, content_type, size, attach) {
361   XBT_DEBUG("Create resource with Bconnection '%f' Bread '%f' Bwrite '%f' and Size '%llu'", bconnection, bread, bwrite, size);
362 }
363
364 StorageActionPtr StorageN11::ls(const char* path)
365 {
366   StorageActionPtr action = new StorageN11Action(getModel(), 0, getState() != SURF_RESOURCE_ON, this, LS);
367
368   action->p_lsDict = NULL;
369   xbt_dict_t ls_dict = xbt_dict_new_homogeneous(xbt_free);
370
371   char* key;
372   sg_size_t size = 0;
373   xbt_dict_cursor_t cursor = NULL;
374
375   xbt_dynar_t dyn = NULL;
376   char* file = NULL;
377
378   // for each file in the storage content
379   xbt_dict_foreach(p_content,cursor,key,size){
380     // Search if file start with the prefix 'path'
381     if(xbt_str_start_with(key,path)){
382       file = &key[strlen(path)];
383
384       // Split file with '/'
385       dyn = xbt_str_split(file,"/");
386       file = xbt_dynar_get_as(dyn,0,char*);
387
388       // file
389       if(xbt_dynar_length(dyn) == 1){
390         sg_size_t *psize = xbt_new(sg_size_t, 1);
391         *psize=size;
392         xbt_dict_set(ls_dict, file, psize, NULL);
393       }
394       // Directory
395       else
396       {
397         // if directory does not exist yet in the dictionary
398         if(!xbt_dict_get_or_null(ls_dict,file))
399           xbt_dict_set(ls_dict,file,NULL,NULL);
400       }
401       xbt_dynar_free(&dyn);
402     }
403   }
404
405   action->p_lsDict = ls_dict;
406   return action;
407 }
408
409 StorageActionPtr StorageN11::open(const char* mount, char* path)
410 {
411   XBT_DEBUG("\tOpen file '%s'",path);
412
413   sg_size_t size, *psize;
414   psize = (sg_size_t*) xbt_dict_get_or_null(p_content, path);
415   // if file does not exist create an empty file
416   if(psize)
417     size = *psize;
418   else {
419         psize = xbt_new(sg_size_t,1);
420     size = 0;
421     *psize = size;
422     xbt_dict_set(p_content, path, psize, NULL);
423     XBT_DEBUG("File '%s' was not found, file created.",path);
424   }
425   surf_file_t file = xbt_new0(s_surf_file_t,1);
426   file->name = xbt_strdup(path);
427   file->size = size;
428   file->mount = xbt_strdup(mount);
429   file->current_position = 0;
430
431   StorageActionPtr action = new StorageN11Action(getModel(), 0, getState() != SURF_RESOURCE_ON, this, OPEN);
432   action->p_file = file;
433   free(path);
434   return action;
435 }
436
437 StorageActionPtr StorageN11::close(surf_file_t fd)
438 {
439   char *filename = fd->name;
440   XBT_DEBUG("\tClose file '%s' size '%llu'", filename, fd->size);
441   // unref write actions from storage
442   void *_write_action;
443   StorageActionPtr write_action;
444   unsigned int i;
445   xbt_dynar_foreach(p_writeActions, i, _write_action) {
446         write_action = static_cast<StorageActionPtr>(static_cast<ActionPtr>(_write_action));
447     if ((write_action->p_file) == fd) {
448       xbt_dynar_cursor_rm(p_writeActions, &i);
449       write_action->unref();
450     }
451   }
452   free(fd->name);
453   free(fd->mount);
454   xbt_free(fd);
455   StorageActionPtr action = new StorageN11Action(getModel(), 0, getState() != SURF_RESOURCE_ON, this, CLOSE);
456   return action;
457 }
458
459 StorageActionPtr StorageN11::read(surf_file_t fd, sg_size_t size)
460 {
461   if(size > fd->size){
462     size = fd->size;
463     fd->current_position = fd->size;
464   }
465   else
466         fd->current_position += size;
467
468   StorageActionPtr action = new StorageN11Action(getModel(), size, getState() != SURF_RESOURCE_ON, this, READ);
469   return action;
470 }
471
472 StorageActionPtr StorageN11::write(surf_file_t fd, sg_size_t size)
473 {
474   char *filename = fd->name;
475   XBT_DEBUG("\tWrite file '%s' size '%llu/%llu'",filename,size,fd->size);
476
477   StorageActionPtr action = new StorageN11Action(getModel(), size, getState() != SURF_RESOURCE_ON, this, WRITE);
478   action->p_file = fd;
479   fd->current_position += size;
480   // If the storage is full
481   if(m_usedSize==m_size) {
482     action->setState(SURF_ACTION_FAILED);
483   }
484   return action;
485 }
486
487 void StorageN11::rename(const char *src, const char *dest)
488 {
489   sg_size_t *psize, *new_psize;
490   psize = (sg_size_t*) xbt_dict_get_or_null(p_content,src);
491   new_psize = xbt_new(sg_size_t, 1);
492   *new_psize = *psize;
493   if (psize){// src file exists
494     xbt_dict_remove(p_content, src);
495     xbt_dict_set(p_content, dest, new_psize,NULL);
496     XBT_DEBUG("Change file name from %s to %s, size '%llu'",src, dest, *psize);
497   }
498   else
499     XBT_DEBUG("File %s doesn't exist",src);
500 }
501
502 /**********
503  * Action *
504  **********/
505
506 StorageN11Action::StorageN11Action(ModelPtr model, double cost, bool failed, StoragePtr storage, e_surf_action_storage_type_t type)
507   : StorageAction(model, cost, failed,
508                       lmm_variable_new(model->getMaxminSystem(), this, 1.0, -1.0 , 3),
509                       storage, type) {
510   XBT_IN("(%s,%g", storage->getName(), cost);
511
512   // Must be less than the max bandwidth for all actions
513   lmm_expand(model->getMaxminSystem(), storage->getConstraint(), getVariable(), 1.0);
514   switch(type) {
515   case OPEN:
516   case CLOSE:
517   case STAT:
518   case LS:
519     break;
520   case READ:
521     lmm_expand(model->getMaxminSystem(), storage->p_constraintRead,
522                    getVariable(), 1.0);
523     break;
524   case WRITE:
525     lmm_expand(model->getMaxminSystem(), storage->p_constraintWrite,
526                    getVariable(), 1.0);
527     ActionPtr action = this;
528     xbt_dynar_push(storage->p_writeActions, &action);
529     ref();
530     break;
531   }
532   XBT_OUT();
533 }
534
535 int StorageN11Action::unref()
536 {
537   m_refcount--;
538   if (!m_refcount) {
539         if (actionHook::is_linked())
540           p_stateSet->erase(p_stateSet->iterator_to(*this));
541     if (getVariable())
542       lmm_variable_free(getModel()->getMaxminSystem(), getVariable());
543 #ifdef HAVE_TRACING
544     xbt_free(getCategory());
545 #endif
546     delete this;
547     return 1;
548   }
549   return 0;
550 }
551
552 void StorageN11Action::cancel()
553 {
554   setState(SURF_ACTION_FAILED);
555   return;
556 }
557
558 void StorageN11Action::suspend()
559 {
560   XBT_IN("(%p)", this);
561   if (m_suspended != 2) {
562     lmm_update_variable_weight(getModel()->getMaxminSystem(),
563                                getVariable(),
564                                0.0);
565     m_suspended = 1;
566   }
567   XBT_OUT();
568 }
569
570 void StorageN11Action::resume()
571 {
572   THROW_UNIMPLEMENTED;
573 }
574
575 bool StorageN11Action::isSuspended()
576 {
577   return m_suspended == 1;
578 }
579
580 void StorageN11Action::setMaxDuration(double /*duration*/)
581 {
582   THROW_UNIMPLEMENTED;
583 }
584
585 void StorageN11Action::setPriority(double /*priority*/)
586 {
587   THROW_UNIMPLEMENTED;
588 }
589