- Comm.detach(): start and forget about asynchronous emission
- this_actor::send(mailbox) is now mailbox->put()
+ SURF
+ - Unused option network/sender-gap is removed.
+
+ MSG
+ - Deprecate MSG_task_isend_with_matching(): This unused feature
+ really complicates our internals. Will be removed in v3.20.
+
SimGrid (3.16) Released June 22. 2017.
The Blooming Spring Release: developments are budding.
SMPI
- New algorithm to privatize globals: dlopen, with dynamic loading tricks
- New option: smpi/keep-temps to not cleanup temp files
- - New option : smpi/shared-malloc-blocksize . Relevant only when global shared
+ - New option : smpi/shared-malloc-blocksize . Relevant only when global shared
mallocs mode is used, allows to change the size of the fake file used
(default 1MB), to potentially limit the number of mappings for large runs.
- Support for sparse privatized malloc with SMPI_PARTIAL_SHARED_MALLOC()
MSG
- New: MSG_process_yield(). Stop and yield to other processes.
- - New: MSG_process_daemon(). Daemon processes are automatically killed
+ - New: MSG_process_daemon(). Daemon processes are automatically killed
when the last non-daemon process terminates
- New: MSG_process_ref/unref(). Fiddle with the process refcounting.
- Renamed MSG_energy_plugin_init() -> MSG_host_energy_plugin_init()
- Kill the obscure NativeException. Nobody want to survive the issues
it denotes, so use JniException that is a RuntimeException (not to
be caught explicitly).
- - Partial bug fix in initialization. SimGrid flags on command line were
- consumed at C level but stayed in the original Java String[] args.
+ - Partial bug fix in initialization. SimGrid flags on command line were
+ consumed at C level but stayed in the original Java String[] args.
This could mess users' args[i] if SG flags were not put at the end of
the command line.
- The SimGrid flags are now removed from the Java arguments. However,
+ The SimGrid flags are now removed from the Java arguments. However,
the number of arguments REMAINS UNCHANGED. It is then UNSAFE to test
- if args.length is greater than the number of YOUR OWN ARGUMENTS.
+ if args.length is greater than the number of YOUR OWN ARGUMENTS.
It might be if you have --log or --cfg flags in the command line.
- Fix numerous memleaks all around the place. In particular, around VMs.
SimDag
- Backwards Compatibility breaks
- - The SD_route_get_list and SD_route_get_size functions have been removed.
- They are replaced by the unique
+ - The SD_route_get_list and SD_route_get_size functions have been removed.
+ They are replaced by the unique
void sg_host_route(sg_host_t src, sg_host_t dst, xbt_dynar_t links)
- The route from src to dst is built in the links dynar whose size can
+ The route from src to dst is built in the links dynar whose size can
be obtained with xbt_dynar_length.
- The SD_route_bandwidth and SD_route_latency functions have been replaced by
- sg_host_route_bandwidth and sg_host_route_latency.
+ sg_host_route_bandwidth and sg_host_route_latency.
Macros ensure the backwards compatibility, but you should fix your code
SMPI
- - Major C++ rewrite ongoing (SMPI used to be C compiled in C++).
+ - Major C++ rewrite ongoing (SMPI used to be C compiled in C++).
This can break codes using SMPI internals (from private.h instead of the public smpi.h).
- - Bump our claim of support from MPI 1.1 to MPI 2.2.
+ - Bump our claim of support from MPI 1.1 to MPI 2.2.
We don't support 100% of it, but it should be enough. Contact us if not.
- - MPI_Win_lock/unlock, MPI_Get_accumulate support added (as for all RMA, implementation is
+ - MPI_Win_lock/unlock, MPI_Get_accumulate support added (as for all RMA, implementation is
naive and probably inaccurate)
- New algorithm for SMPI_SHARED_MALLOC: global, where all blocks are
mapped onto a unique small file using some system magic.
SMPI
* Privatization now works on FreeBSD
-
+
* Privatization is now activated by default in smpirun
You can turn it off with -no-privatize if something goes wrong.
* Call-location tracing for SMPI.
- You can add the exact location (filename / linenumber) of an MPI call to
- your trace files and slow down or speed up the simulation between two
+ You can add the exact location (filename / linenumber) of an MPI call to
+ your trace files and slow down or speed up the simulation between two
consecutive calls by using an adjustment file (see the documentation).
* Fixed computation of timings for MPI_Send, MPI_Recv & possibly also others
* smpi/cpu-threshold:-1 should become smpi/simulate-computation:no
smpi/running-power is renamed to smpi/host-speed
- * smpi/grow-injected-times option to enable or disable multiplication of the
- timings injected in MPI_Iprobe, or MPI_Test. Enabled by default, which can
+ * smpi/grow-injected-times option to enable or disable multiplication of the
+ timings injected in MPI_Iprobe, or MPI_Test. Enabled by default, which can
make simulation less precise (but also much faster).
* smpirun script should be (much) faster for large deployments.
- Ignored parameter of vm_create: core_nb, disk_path and disk_size.
- Unimplemented save/restore methods
* MSG_as_router_get_property_value() was redundent with
- MSG_environment_as_get_property_value().
+ MSG_environment_as_get_property_value().
- Removed MSG_as_router_*propert*() functions
- Added MSG_environment_as_set_property_value() for consistency
- * xbt heterogeneous dictionnaries (created with xbt_dict_new()).
- Well, they are still there for now, but deprecated with a warning.
+ * xbt heterogeneous dictionnaries (created with xbt_dict_new()).
+ Well, they are still there for now, but deprecated with a warning.
Please switch to xbt_dict_new_homogeneous() before this is removed
for real.
* Task affinity. Its intended behavior (that was very badly tested
would have hoped here.
* xbt_os_sem_get_value: unused internally, deprecated on OS X El Capitan
* Option network/coordinates is now useless and should be dropped.
-
- Storage:
+
+ Storage:
* Added option storage/max_file_descriptors to allow more than 1024 files opened
SURF:
XML:
* Add Exa- and Peta- units such as EiB, EB, Eib, Eb for size, and
- EiBps, EBps, Eibps, Ebps for bandwidth.
+ EiBps, EBps, Eibps, Ebps for bandwidth.
They may become useful to some lucky ones.
-
+
Java:
* New functions: msg.Comm.waitAll() and msg.Comm.waitAny()
* ex/app_tokenring: new example, very similar to the MSG Token Ring
* ex/async_waitAll: new example, on asynchronous communications
MSG:
- * Memory usage should be decreased for simulations with a large number
+ * Memory usage should be decreased for simulations with a large number
of processes. This also helps for SMPI.
SimGrid (3.13) stable; urgency=low
The Half Release, a.k.a. the Zealous Easter Trim.
-
+
- We removed half of the lines, that were mostly experimental cruft.
v3.12 lasted 286000 lines of code, v3.13 is only 142000 lines
(not counting blanks and comments -- according to openhub.net)
- The internals are now compiled in C++ (and will soon be clean C++)
- - We removed 75 klines of XML, 12 klines of Java, 5 klines of cmake,
+ - We removed 75 klines of XML, 12 klines of Java, 5 klines of cmake,
59 klines of C, etc. We added only 29 klines of C++ in replacement.
-
+
* Backwards Compatibility breaks
- Removed Lua simulation bindings (switch to C or Java for that).
Lua can still be used to describe platforms
way, with filter function.
- MSG_task_listen_from_host
- MSG_mailbox_get_count_host_waiting_tasks
- - MSG_mailbox_put_with_timeout was removed.
+ - MSG_mailbox_put_with_timeout was removed.
Please use MSG_task_send_with_timeout instead.
- - In SimDag
- - the SD_application_reinit function was removed. It has been a noop for a while.
- - The ACCESS_MODE of SD_workstation has been removed. This feature was not really usable and should soon be
+ - In SimDag
+ - the SD_application_reinit function was removed. It has been a noop for a while.
+ - The ACCESS_MODE of SD_workstation has been removed. This feature was not really usable and should soon be
replaced by a more flexible mechanism.
- The following functions thus do not exist anymore
- SD_workstation_get_access_mode
- SD_route_get_communication_time => SG_route_get_latency() + amount / SD_route_get_bandwidth()
- SD_workstation_get_computation_time => amount / sg_host_speed()
- In Java
- - VM.setBound(int load) is now VM.setBound(double bound) to meet the MSG semantics.
+ - VM.setBound(int load) is now VM.setBound(double bound) to meet the MSG semantics.
Use VM.getSpeed()*load/100 for the legacy behavior.
- In CMake
- option enable_tracing was removed. It was not doing anything for a while.
And the same in bits: 'Tibps', 'Gibps', 'Mibps', 'Kibps', 'Tbps', 'Gbps', 'Mbps', 'kbps', 'bps'
- latency. Default: 's' second. Also defined:
'w' week, 'd' day, 'h' hour, 'm' minute, 'ms' millisecond, 'us' microsecond, 'ns' nanosecond, 'ps' picosecond
-
+
* bin/simgrid_update_xml can upgrade your files automatically (won't convert unit-less values)
tools/sg_xml_unit_converter.py may help (but it's just a warning and will probably ever be).
MSG
* The examples were completely reorganized (in C and Java), for your browsing pleasure.
- * Kill all deprecated functions (the ones you had when declaring MSG_DEPRECATED).
+ * Kill all deprecated functions (the ones you had when declaring MSG_DEPRECATED).
They were deprecated since a few years, and probably did not even compile anymore.
SimDag
- * The API has been profoundly modified to directly use the core objects instead of redefining its own.
+ * The API has been profoundly modified to directly use the core objects instead of redefining its own.
SD_Workstation_t and SD_link_t are now sg_host_t and sg_link_t respectively.
- Some functions have also been renamed for consistency. Backward compatibility is maintained, but users are
- encouraged to update their codes. A list of the modified functions can be found at the end of
+ Some functions have also been renamed for consistency. Backward compatibility is maintained, but users are
+ encouraged to update their codes. A list of the modified functions can be found at the end of
include/simgrid/simdag.h
Simix
* refactoring and cleanup of the code;
* ongoing process to cleanly separate the model-checking algorithms
from the code model-checking support.
-
+
-- Wed Apr 27 21:00:53 CEST 2016 Da SimGrid team <simgrid-devel@lists.gforge.inria.fr>
SimGrid (3.12) stable; urgency=low
@CMAKE_HOME_DIRECTORY@/include/simgrid \
@CMAKE_HOME_DIRECTORY@/include/simgrid/s4u \
@CMAKE_HOME_DIRECTORY@/src/include/surf \
+ @CMAKE_HOME_DIRECTORY@/src/include/xbt \
@CMAKE_HOME_DIRECTORY@/src/msg/ \
@CMAKE_HOME_DIRECTORY@/src/kernel/ \
@CMAKE_HOME_DIRECTORY@/src/kernel/activity/ \
- \c network/maxmin-selective-update: \ref options_model_optim
- \c network/model: \ref options_model_select
- \c network/optim: \ref options_model_optim
-- \c network/sender_gap: \ref options_model_network_sendergap
- \c network/TCP-gamma: \ref options_model_network_gamma
- \c network/weight-S: \ref options_model_network_coefs
Note that with the default host model this option is activated by default.
-\subsubsection options_model_network_sendergap Simulating sender gap
-
-(this configuration item is experimental and may change or disapear)
-
-It is possible to specify a timing gap between consecutive emission on
-the same network card through the \b network/sender-gap item. This
-is still under investigation as of writting, and the default value is
-to wait 10 microseconds (1e-5 seconds) between emissions.
-
\subsubsection options_model_network_asyncsend Simulating asyncronous send
(this configuration item is experimental and may change or disapear)
set_target_properties(network-ns3 PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/network-ns3)
endif()
-add_executable (bittorrent app-bittorrent/bittorrent.c app-bittorrent/messages.c app-bittorrent/peer.c app-bittorrent/tracker.c app-bittorrent/connection.c)
-target_link_libraries(bittorrent simgrid)
-set_target_properties(bittorrent PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/app-bittorrent)
-foreach (file bittorrent connection messages peer tracker)
- set(examples_src ${examples_src} ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/${file}.c ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/${file}.h)
-endforeach()
-
add_executable (chainsend app-chainsend/chainsend.c app-chainsend/iterator.c app-chainsend/common.c app-chainsend/messages.c app-chainsend/broadcaster.c app-chainsend/peer.c)
target_link_libraries(chainsend simgrid)
set_target_properties(chainsend PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/app-chainsend)
set(examples_src ${examples_src} ${CMAKE_CURRENT_SOURCE_DIR}/dht-kademlia/${file}.c ${CMAKE_CURRENT_SOURCE_DIR}/dht-kademlia/${file}.h)
endforeach()
-foreach (file app-bittorrent app-chainsend app-masterworker app-pingpong async-wait
+foreach (file app-chainsend app-masterworker app-pingpong async-wait
async-waitall async-waitany dht-kademlia dht-pastry io-remote platform-properties process-yield
task-priority)
set(xml_files ${xml_files} ${CMAKE_CURRENT_SOURCE_DIR}/${file}/${file}_d.xml)
endforeach()
set(txt_files ${txt_files} ${CMAKE_CURRENT_SOURCE_DIR}/README.doc PARENT_SCOPE)
-set(bin_files ${bin_files} ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/generate.py
- ${CMAKE_CURRENT_SOURCE_DIR}/dht-kademlia/generate.py
+set(bin_files ${bin_files} ${CMAKE_CURRENT_SOURCE_DIR}/dht-kademlia/generate.py
${CMAKE_CURRENT_SOURCE_DIR}/dht-pastry/generate.py PARENT_SCOPE)
set(examples_src ${examples_src} ${CMAKE_CURRENT_SOURCE_DIR}/app-chainsend/chainsend.c
${CMAKE_CURRENT_SOURCE_DIR}/dht-kademlia/common.h
${CMAKE_CURRENT_SOURCE_DIR}/network-ns3/network-ns3.c PARENT_SCOPE)
-set(tesh_files ${tesh_files} ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/app-bittorrent.tesh
- ${CMAKE_CURRENT_SOURCE_DIR}/app-chainsend/app-chainsend.tesh
+set(tesh_files ${tesh_files} ${CMAKE_CURRENT_SOURCE_DIR}/app-chainsend/app-chainsend.tesh
${CMAKE_CURRENT_SOURCE_DIR}/dht-kademlia/dht-kademlia.tesh
${CMAKE_CURRENT_SOURCE_DIR}/app-masterworker/app-masterworker-multicore.tesh
${CMAKE_CURRENT_SOURCE_DIR}/app-masterworker/app-masterworker-vivaldi.tesh
${CMAKE_CURRENT_SOURCE_DIR}/process-startkilltime/start_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/process-startkilltime/start_kill_d.xml PARENT_SCOPE)
-foreach(x app-bittorrent app-chainsend app-masterworker app-pingpong app-token-ring
+foreach(x app-chainsend app-masterworker app-pingpong app-token-ring
async-wait async-waitall async-waitany cloud-capping cloud-masterworker cloud-migration cloud-simple
cloud-two-tasks dht-pastry dht-kademlia platform-failures io-file io-remote io-storage task-priority
process-create process-daemon process-join process-kill process-migration process-startkilltime process-suspend process-yield
ADD_TESH(msg-trace-${x} --setenv bindir=${CMAKE_BINARY_DIR}/examples/msg/trace-${x} --setenv srcdir=${CMAKE_HOME_DIRECTORY}/examples/platforms --cd ${CMAKE_HOME_DIRECTORY}/examples/msg/trace-${x} trace-${x}.tesh)
endforeach()
-ADD_TESH_FACTORIES(msg-app-bittorrent-parallel "thread;ucontext;raw" --cfg contexts/nthreads:4 ${CONTEXTS_SYNCHRO} --setenv bindir=${CMAKE_BINARY_DIR}/examples/msg/app-bittorrent --setenv srcdir=${CMAKE_HOME_DIRECTORY}/examples/platforms --cd ${CMAKE_HOME_DIRECTORY}/examples/msg/app-bittorrent app-bittorrent.tesh)
ADD_TESH_FACTORIES(msg-dht-kademlia-parallel "thread;ucontext;raw" --cfg contexts/nthreads:4 ${CONTEXTS_SYNCHRO} --setenv bindir=${CMAKE_BINARY_DIR}/examples/msg/dht-kademlia --setenv srcdir=${CMAKE_HOME_DIRECTORY}/examples/platforms --cd ${CMAKE_HOME_DIRECTORY}/examples/msg/dht-kademlia dht-kademlia.tesh)
ADD_TESH_FACTORIES(msg-energy-pstate-ptask "thread;ucontext;raw;boost" --cfg host/model:ptask_L07 --log xbt_cfg.threshold:critical --setenv srcdir=${CMAKE_HOME_DIRECTORY}/examples/msg --cd ${CMAKE_BINARY_DIR}/examples/msg ${CMAKE_HOME_DIRECTORY}/examples/msg/energy-pstate/energy-pstate.tesh)
ADD_TESH_FACTORIES(msg-energy-consumption-ptask "thread;ucontext;raw;boost" --cfg host/model:ptask_L07 --log xbt_cfg.threshold:critical --setenv srcdir=${CMAKE_HOME_DIRECTORY}/examples/msg --cd ${CMAKE_BINARY_DIR}/examples/msg ${CMAKE_HOME_DIRECTORY}/examples/msg/energy-consumption/energy-consumption.tesh)
+++ /dev/null
-/* Copyright (c) 2012-2014. The SimGrid Team.
- * All rights reserved. */
-
-/* This program is free software; you can redistribute it and/or modify it
- * under the terms of the license (GNU LGPL) which comes with this package. */
-
-#include "messages.h"
-#include "bittorrent.h"
-
-XBT_LOG_NEW_DEFAULT_CATEGORY(msg_messages, "Messages specific for the message factory");
-
-#define BITS_TO_BYTES(x) ((x / 8) + (x % 8) ? 1 : 0)
-
-/** @brief Build a new empty message
- * @param type type of the message
- * @param issuer_host_name hostname of the issuer, for debugging purposes
- * @param mailbox mailbox where the peer should answer
- * @param peer_id id of the issuer
- * @param size message size in bytes
- */
-msg_task_t task_message_new(e_message_type type, const char *issuer_host_name, const char *mailbox, int peer_id,
- int size)
-{
- message_t message = xbt_new(s_message_t, 1);
- message->issuer_host_name = issuer_host_name;
- message->peer_id = peer_id;
- message->mailbox = mailbox;
- message->type = type;
- msg_task_t task = MSG_task_create(NULL, 0, size, message);
- XBT_DEBUG("type: %d size: %d", (int) type, size);
- return task;
-}
-
-/** Builds a message containing an index. */
-msg_task_t task_message_index_new(e_message_type type, const char *issuer_host_name, const char *mailbox, int peer_id,
- int index, int varsize)
-{
- msg_task_t task = task_message_new(type, issuer_host_name, mailbox, peer_id, task_message_size(type) + varsize);
- message_t message = MSG_task_get_data(task);
- message->index = index;
- return task;
-}
-
-msg_task_t task_message_bitfield_new(const char *issuer_host_name, const char *mailbox, int peer_id,
- unsigned int bitfield, int bitfield_size)
-{
- msg_task_t task = task_message_new(MESSAGE_BITFIELD, issuer_host_name, mailbox, peer_id,
- task_message_size(MESSAGE_BITFIELD) + BITS_TO_BYTES(bitfield_size));
- message_t message = MSG_task_get_data(task);
- message->bitfield = bitfield;
- return task;
-}
-
-msg_task_t task_message_request_new(const char *issuer_host_name, const char *mailbox, int peer_id, int index,
- int block_index, int block_length)
-{
- msg_task_t task = task_message_index_new(MESSAGE_REQUEST, issuer_host_name, mailbox, peer_id, index, 0);
- message_t message = MSG_task_get_data(task);
- message->block_index = block_index;
- message->block_length = block_length;
- return task;
-}
-
-msg_task_t task_message_piece_new(const char *issuer_host_name, const char *mailbox, int peer_id, int index,
- int block_index, int block_length, int block_size)
-{
- msg_task_t task = task_message_index_new(MESSAGE_PIECE, issuer_host_name, mailbox, peer_id, index,
- block_length * block_size);
- message_t message = MSG_task_get_data(task);
- message->block_index = block_index;
- message->block_length = block_length;
- return task;
-}
-
-void task_message_free(void *task)
-{
- message_t message = MSG_task_get_data(task);
- xbt_free(message);
- MSG_task_destroy(task);
-}
-
-int task_message_size(e_message_type type)
-{
- int size = 0;
- switch (type) {
- case MESSAGE_HANDSHAKE:
- size = MESSAGE_HANDSHAKE_SIZE;
- break;
- case MESSAGE_CHOKE:
- size = MESSAGE_CHOKE_SIZE;
- break;
- case MESSAGE_UNCHOKE:
- size = MESSAGE_UNCHOKE_SIZE;
- break;
- case MESSAGE_INTERESTED:
- case MESSAGE_NOTINTERESTED:
- size = MESSAGE_INTERESTED_SIZE;
- break;
- case MESSAGE_HAVE:
- size = MESSAGE_HAVE_SIZE;
- break;
- case MESSAGE_BITFIELD:
- size = MESSAGE_BITFIELD_SIZE;
- break;
- case MESSAGE_REQUEST:
- size = MESSAGE_REQUEST_SIZE;
- break;
- case MESSAGE_PIECE:
- size = MESSAGE_PIECE_SIZE;
- break;
- case MESSAGE_CANCEL:
- size = MESSAGE_CANCEL_SIZE;
- break;
- default:
- THROW_IMPOSSIBLE;
- }
- return size;
-}
+++ /dev/null
-/* Copyright (c) 2012-2014. The SimGrid Team.
- * All rights reserved. */
-
-/* This program is free software; you can redistribute it and/or modify it
- * under the terms of the license (GNU LGPL) which comes with this package. */
-
-#ifndef BITTORRENT_TRACKER_H_
-#define BITTORRENT_TRACKER_H_
-#include <xbt/dynar.h>
-#include "bittorrent.h"
-/**
- * Tracker main function
- */
-int tracker(int argc, char *argv[]);
-/**
- * Task types exchanged between a node and the tracker
- */
-typedef enum {
- TRACKER_TASK_QUERY,
- TRACKER_TASK_ANSWER
-} e_tracker_task_type_t;
-/**
- * Tasks exchanged between a tracker and peers.
- */
-typedef struct s_tracker_task_data {
- e_tracker_task_type_t type; //type of the task
- const char *mailbox; //mailbox where the tracker should answer
- const char *issuer_host_name; //hostname, for debug purposes
- //Query data
- int peer_id; //peer id
- int uploaded; //how much the peer has already uploaded
- int downloaded; //how much the peer has downloaded
- int left; //how much the peer has left
- //Answer data
- int interval; //how often the peer should contact the tracker (unused for now)
- xbt_dynar_t peers; //the peer list the peer has asked for.
-} s_tracker_task_data_t;
-typedef s_tracker_task_data_t *tracker_task_data_t;
-
-tracker_task_data_t tracker_task_data_new(const char *issuer_host_name, const char *mailbox, int peer_id,
- int uploaded, int downloaded, int left);
-void tracker_task_data_free(tracker_task_data_t task);
-
-int is_in_list(xbt_dynar_t peers, int id);
-#endif /* BITTORRENT_TRACKER_H */
> [0.000000] [xbt_cfg/INFO] Configuration change: Set 'viva/uncategorized' to 'categories.uncat.plist'
$ $SG_TEST_EXENV cat categories.cat.plist
-> node = ("HOST","LINK");
-> edge = ("0-LINK4-LINK4","0-HOST1-LINK4","0-LINK4-HOST1");
+> node = ("LINK","HOST");
+> edge = ("0-LINK4-LINK4","0-LINK4-HOST1","0-HOST1-LINK4");
>
> host = {
> type = "square";
> size = "power";
-> values = ("pdata","pcompute","pfinalize","prequest");
+> values = ("pcompute","pdata","pfinalize","prequest");
> };
> link = {
> type = "rhombus";
> size = "bandwidth";
-> values = ("bdata","bcompute","bfinalize","brequest");
+> values = ("bcompute","bdata","bfinalize","brequest");
> };
$ $SG_TEST_EXENV cat categories.uncat.plist
-> node = ("HOST","LINK");
-> edge = ("0-LINK4-LINK4","0-HOST1-LINK4","0-LINK4-HOST1");
+> node = ("LINK","HOST");
+> edge = ("0-LINK4-LINK4","0-LINK4-HOST1","0-HOST1-LINK4");
>
> host = {
> type = "square";
> [0.004078] [msg_test/INFO] HDD_utilization
> [0.004078] [msg_test/INFO] Declared link variables:
> [0.004078] [msg_test/INFO] Node types in the trace:
-> [0.004078] [msg_test/INFO] LINK
> [0.004078] [msg_test/INFO] HOST
+> [0.004078] [msg_test/INFO] LINK
> [0.004078] [msg_test/INFO] Node types in the trace:
-> [0.004078] [msg_test/INFO] 0-LINK3-LINK3
> [0.004078] [msg_test/INFO] 0-HOST1-LINK3
> [0.004078] [msg_test/INFO] 0-LINK3-HOST1
+> [0.004078] [msg_test/INFO] 0-LINK3-LINK3
$ rm -f viva_graph.plist
> [0.000000] [xbt_cfg/INFO] Configuration change: Set 'viva/categorized' to 'trace-masterworker.cat.plist'
> [0.000000] [xbt_cfg/INFO] Configuration change: Set 'viva/uncategorized' to 'trace-masterworker.uncat.plist'
> [4.214821] [msg_trace_masterworker/INFO] Declared tracing categories:
-> [4.214821] [msg_trace_masterworker/INFO] request
+> [4.214821] [msg_trace_masterworker/INFO] compute
> [4.214821] [msg_trace_masterworker/INFO] finalize
> [4.214821] [msg_trace_masterworker/INFO] report
-> [4.214821] [msg_trace_masterworker/INFO] compute
+> [4.214821] [msg_trace_masterworker/INFO] request
> [4.214821] [msg_trace_masterworker/INFO] Declared marks:
> [4.214821] [msg_trace_masterworker/INFO] msmark
> [0.000000] [xbt_cfg/INFO] Configuration change: Set 'viva/categorized' to 'trace-masterworker.cat.plist'
> [0.000000] [xbt_cfg/INFO] Configuration change: Set 'viva/uncategorized' to 'trace-masterworker.uncat.plist'
> [4.214821] [msg_trace_masterworker/INFO] Declared tracing categories:
-> [4.214821] [msg_trace_masterworker/INFO] request
+> [4.214821] [msg_trace_masterworker/INFO] compute
> [4.214821] [msg_trace_masterworker/INFO] finalize
> [4.214821] [msg_trace_masterworker/INFO] report
-> [4.214821] [msg_trace_masterworker/INFO] compute
+> [4.214821] [msg_trace_masterworker/INFO] request
> [4.214821] [msg_trace_masterworker/INFO] Declared marks:
> [4.214821] [msg_trace_masterworker/INFO] msmark
> [0.000000] [xbt_cfg/INFO] Configuration change: Set 'viva/categorized' to 'trace-masterworker.cat.plist'
> [0.000000] [xbt_cfg/INFO] Configuration change: Set 'viva/uncategorized' to 'trace-masterworker.uncat.plist'
> [4.214821] [msg_trace_masterworker/INFO] Declared tracing categories:
-> [4.214821] [msg_trace_masterworker/INFO] request
+> [4.214821] [msg_trace_masterworker/INFO] compute
> [4.214821] [msg_trace_masterworker/INFO] finalize
> [4.214821] [msg_trace_masterworker/INFO] report
-> [4.214821] [msg_trace_masterworker/INFO] compute
+> [4.214821] [msg_trace_masterworker/INFO] request
> [4.214821] [msg_trace_masterworker/INFO] Declared marks:
> [4.214821] [msg_trace_masterworker/INFO] msmark
$ $SG_TEST_EXENV cat trace-masterworker.uncat.plist
-> node = ("HOST","LINK");
+> node = ("LINK","HOST");
> edge = ("0-LINK13-LINK13","0-LINK13-HOST1","0-HOST1-LINK13");
->
+>
> host = {
> type = "square";
> size = "power";
> };
$ $SG_TEST_EXENV cat trace-masterworker.cat.plist
-> node = ("HOST","LINK");
+> node = ("LINK","HOST");
> edge = ("0-LINK13-LINK13","0-LINK13-HOST1","0-HOST1-LINK13");
->
+>
> host = {
> type = "square";
> size = "power";
-> values = ("pcompute","preport","pfinalize","prequest");
+> values = ("pcompute","pfinalize","preport","prequest");
> };
> link = {
> type = "rhombus";
> size = "bandwidth";
-> values = ("bcompute","breport","bfinalize","brequest");
+> values = ("bcompute","bfinalize","breport","brequest");
> };
$ rm -rf trace-masterworker.trace trace-masterworker.cat.plist trace-masterworker.uncat.plist
endforeach()
set(examples_src ${examples_src} ${CMAKE_CURRENT_SOURCE_DIR}/dht-chord/s4u_dht-chord.hpp)
+add_executable (s4u_bittorrent app-bittorrent/s4u_bittorrent.cpp app-bittorrent/s4u_peer.cpp
+ app-bittorrent/s4u_tracker.cpp)
+target_link_libraries(s4u_bittorrent simgrid)
+set_target_properties(s4u_bittorrent PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/app-bittorrent)
+foreach (file s4u_bittorrent s4u_peer s4u_tracker)
+ set(examples_src ${examples_src} ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/${file}.cpp
+ ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/${file}.hpp)
+endforeach()
+
set(examples_src ${examples_src} PARENT_SCOPE)
-set(tesh_files ${tesh_files} ${CMAKE_CURRENT_SOURCE_DIR}/dht-chord/s4u_dht-chord.tesh PARENT_SCOPE)
+set(tesh_files ${tesh_files} ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/s4u_app-bittorrent.tesh
+ ${CMAKE_CURRENT_SOURCE_DIR}/dht-chord/s4u_dht-chord.tesh PARENT_SCOPE)
set(xml_files ${xml_files} ${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/s4u_actions-comm_split_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/s4u_actions-comm_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/actions-storage/s4u_actions-storage_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/actor-create/s4u_actor-create_d.xml
+ ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/s4u_app-bittorrent_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/app-masterworker/s4u_app-masterworker_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/dht-chord/s4u_dht-chord_d.xml PARENT_SCOPE)
set(txt_files ${txt_files} ${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/s4u_actions-comm_split_p0.txt
${CMAKE_CURRENT_SOURCE_DIR}/README.doc PARENT_SCOPE)
foreach(example actions-comm actions-storage actor-create actor-daemon actor-kill actor-migration actor-suspend
- app-masterworker app-pingpong app-token-ring dht-chord plugin-hostload io mutex )
+ app-bittorrent app-masterworker app-pingpong app-token-ring dht-chord plugin-hostload io mutex )
ADD_TESH_FACTORIES(s4u-${example} "thread;ucontext;raw;boost" --setenv bindir=${CMAKE_CURRENT_BINARY_DIR}/${example} --setenv srcdir=${CMAKE_HOME_DIRECTORY}/examples/platforms --cd ${CMAKE_HOME_DIRECTORY}/examples/s4u/${example} s4u_${example}.tesh)
endforeach()
--- /dev/null
+#! ./tesh
+
+p Testing the Bittorrent implementation with MSG
+
+! timeout 10
+! output sort 19
+$ $SG_TEST_EXENV ${bindir:=.}/s4u_bittorrent ${srcdir:=.}/cluster.xml ${srcdir:=.}/../s4u/app-bittorrent/s4u_app-bittorrent_d.xml "--log=root.fmt:[%12.6r]%e(%i:%P@%h)%e%m%n"
+> [ 0.000000] (1:tracker@node-0.acme.org) Tracker launched.
+> [ 0.000000] (2:peer@node-1.acme.org) Hi, I'm joining the network with id 2
+> [ 0.000000] (3:peer@node-2.acme.org) Hi, I'm joining the network with id 3
+> [ 0.000000] (4:peer@node-3.acme.org) Hi, I'm joining the network with id 4
+> [ 0.000000] (5:peer@node-4.acme.org) Hi, I'm joining the network with id 5
+> [ 0.000000] (6:peer@node-5.acme.org) Hi, I'm joining the network with id 6
+> [ 0.000000] (7:peer@node-6.acme.org) Hi, I'm joining the network with id 7
+> [ 0.000000] (8:peer@node-7.acme.org) Hi, I'm joining the network with id 8
+> [ 3000.000000] (1:tracker@node-0.acme.org) Tracker is leaving
+> [ 5000.007806] (2:peer@node-1.acme.org) Here is my current status: 1111111111
+> [ 5000.007806] (3:peer@node-2.acme.org) Here is my current status: 1111111111
+> [ 5000.007806] (4:peer@node-3.acme.org) Here is my current status: 1111111111
+> [ 5000.007806] (5:peer@node-4.acme.org) Here is my current status: 1111111111
+> [ 5000.007806] (6:peer@node-5.acme.org) Here is my current status: 1111111111
+> [ 5000.007806] (7:peer@node-6.acme.org) Here is my current status: 1111111111
+> [ 5000.007806] (8:peer@node-7.acme.org) Here is my current status: 1111111111
--- /dev/null
+/* Copyright (c) 2012-2017. The SimGrid Team.
+ * All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#include "s4u_bittorrent.hpp"
+#include "s4u_peer.hpp"
+#include "s4u_tracker.hpp"
+
+simgrid::xbt::Extension<simgrid::s4u::Host, HostBittorrent> HostBittorrent::EXTENSION_ID;
+
+int main(int argc, char* argv[])
+{
+ simgrid::s4u::Engine* e = new simgrid::s4u::Engine(&argc, argv);
+
+ /* Check the arguments */
+ xbt_assert(argc > 2, "Usage: %s platform_file deployment_file", argv[0]);
+
+ e->loadPlatform(argv[1]);
+
+ HostBittorrent::EXTENSION_ID = simgrid::s4u::Host::extension_create<HostBittorrent>();
+
+ std::vector<simgrid::s4u::Host*> list;
+ simgrid::s4u::Engine::getInstance()->getHostList(&list);
+ for (auto host : list)
+ host->extension_set(new HostBittorrent(host));
+
+ e->registerFunction<Tracker>("tracker");
+ e->registerFunction<Peer>("peer");
+ e->loadDeployment(argv[2]);
+
+ e->run();
+
+ delete e;
+ return 0;
+}
--- /dev/null
+/* Copyright (c) 2012-2014, 2016-2017. The SimGrid Team.
+ * All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#ifndef BITTORRENT_BITTORRENT_HPP_
+#define BITTORRENT_BITTORRENT_HPP_
+
+#include <simgrid/s4u.hpp>
+#include <xbt/RngStream.h>
+
+#define MAILBOX_SIZE 40
+#define TRACKER_MAILBOX "tracker_mailbox"
+/** Max number of peers sent by the tracker to clients */
+#define MAXIMUM_PEERS 50
+/** Interval of time where the peer should send a request to the tracker */
+#define TRACKER_QUERY_INTERVAL 1000
+/** Communication size for a task to the tracker */
+#define TRACKER_COMM_SIZE 0.01
+#define GET_PEERS_TIMEOUT 10000
+#define TIMEOUT_MESSAGE 10
+#define TRACKER_RECEIVE_TIMEOUT 10
+/** Number of peers that can be unchocked at a given time */
+#define MAX_UNCHOKED_PEERS 4
+/** Interval between each update of the choked peers */
+#define UPDATE_CHOKED_INTERVAL 30
+/** Number of pieces the peer asks for simultaneously */
+#define MAX_PIECES 1
+
+/** Message sizes
+ * Sizes based on report by A. Legout et al, Understanding BitTorrent: An Experimental Perspective
+ * http://hal.inria.fr/inria-00000156/en
+ */
+#define MESSAGE_HANDSHAKE_SIZE 68
+#define MESSAGE_CHOKE_SIZE 5
+#define MESSAGE_UNCHOKE_SIZE 5
+#define MESSAGE_INTERESTED_SIZE 5
+#define MESSAGE_NOTINTERESTED_SIZE 5
+#define MESSAGE_HAVE_SIZE 9
+#define MESSAGE_BITFIELD_SIZE 5
+#define MESSAGE_REQUEST_SIZE 17
+#define MESSAGE_PIECE_SIZE 13
+#define MESSAGE_CANCEL_SIZE 17
+
+/** Types of messages exchanged between two peers. */
+typedef enum {
+ MESSAGE_HANDSHAKE,
+ MESSAGE_CHOKE,
+ MESSAGE_UNCHOKE,
+ MESSAGE_INTERESTED,
+ MESSAGE_NOTINTERESTED,
+ MESSAGE_HAVE,
+ MESSAGE_BITFIELD,
+ MESSAGE_REQUEST,
+ MESSAGE_PIECE,
+ MESSAGE_CANCEL
+} e_message_type;
+
+class Message {
+public:
+ e_message_type type;
+ int peer_id;
+ simgrid::s4u::MailboxPtr return_mailbox;
+ unsigned int bitfield = 0U;
+ int piece = 0;
+ int block_index = 0;
+ int block_length = 0;
+ Message(e_message_type type, int peer_id, simgrid::s4u::MailboxPtr return_mailbox)
+ : type(type), peer_id(peer_id), return_mailbox(return_mailbox){};
+ Message(e_message_type type, int peer_id, unsigned int bitfield, simgrid::s4u::MailboxPtr return_mailbox)
+ : type(type), peer_id(peer_id), return_mailbox(return_mailbox), bitfield(bitfield){};
+ Message(e_message_type type, int peer_id, simgrid::s4u::MailboxPtr return_mailbox, int piece, int block_index,
+ int block_length)
+ : type(type)
+ , peer_id(peer_id)
+ , return_mailbox(return_mailbox)
+ , piece(piece)
+ , block_index(block_index)
+ , block_length(block_length){};
+ Message(e_message_type type, int peer_id, simgrid::s4u::MailboxPtr return_mailbox, int piece)
+ : type(type), peer_id(peer_id), return_mailbox(return_mailbox), piece(piece){};
+ ~Message() = default;
+};
+
+class HostBittorrent {
+ RngStream stream_;
+ simgrid::s4u::Host* host = nullptr;
+
+public:
+ static simgrid::xbt::Extension<simgrid::s4u::Host, HostBittorrent> EXTENSION_ID;
+
+ explicit HostBittorrent(simgrid::s4u::Host* ptr) : host(ptr)
+ {
+ std::string descr = std::string("RngSream<") + host->getCname() + ">";
+ stream_ = RngStream_CreateStream(descr.c_str());
+ }
+
+ ~HostBittorrent() { RngStream_DeleteStream(&stream_); };
+
+ RngStream getStream() { return stream_; };
+};
+
+#endif /* BITTORRENT_BITTORRENT_HPP_ */
--- /dev/null
+/* Copyright (c) 2012-2017. The SimGrid Team. All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#include <climits>
+#include <xbt/ex.hpp>
+
+#include "s4u_peer.hpp"
+#include "s4u_tracker.hpp"
+
+XBT_LOG_NEW_DEFAULT_CATEGORY(s4u_bt_peer, "Messages specific for the peers");
+
+/*
+ * User parameters for transferred file data. For the test, the default values are :
+ * File size: 10 pieces * 5 blocks/piece * 16384 bytes/block = 819200 bytes
+ */
+#define FILE_PIECES 10UL
+#define PIECES_BLOCKS 5UL
+#define BLOCK_SIZE 16384
+static const unsigned long int FILE_SIZE = FILE_PIECES * PIECES_BLOCKS * BLOCK_SIZE;
+
+/** Number of blocks asked by each request */
+#define BLOCKS_REQUESTED 2
+
+#define ENABLE_END_GAME_MODE 1
+#define SLEEP_DURATION 1
+#define BITS_TO_BYTES(x) (((x) / 8 + (x) % 8) ? 1 : 0)
+
+Peer::Peer(std::vector<std::string> args)
+{
+ // Check arguments
+ xbt_assert(args.size() == 3 || args.size() == 4, "Wrong number of arguments");
+ try {
+ id = std::stoi(args[1]);
+ mailbox_ = simgrid::s4u::Mailbox::byName(std::to_string(id));
+ } catch (std::invalid_argument& ia) {
+ throw std::invalid_argument(std::string("Invalid ID:") + args[1].c_str());
+ }
+
+ try {
+ deadline = std::stod(args[2]);
+ } catch (std::invalid_argument& ia) {
+ throw std::invalid_argument(std::string("Invalid deadline:") + args[2].c_str());
+ }
+ xbt_assert(deadline > 0, "Wrong deadline supplied");
+
+ stream = simgrid::s4u::this_actor::getHost()->extension<HostBittorrent>()->getStream();
+
+ if (args.size() == 4 && args[3] == "1") {
+ bitfield_ = (1U << FILE_PIECES) - 1U;
+ bitfield_blocks = (1ULL << (FILE_PIECES * PIECES_BLOCKS)) - 1ULL;
+ }
+ pieces_count = new short[FILE_PIECES]{0};
+
+ XBT_INFO("Hi, I'm joining the network with id %d", id);
+}
+
+Peer::~Peer()
+{
+ for (auto peer : connected_peers)
+ delete peer.second;
+ delete[] pieces_count;
+}
+
+/** Peer main function */
+void Peer::operator()()
+{
+ // Getting peer data from the tracker.
+ if (getPeersFromTracker()) {
+ XBT_DEBUG("Got %zu peers from the tracker. Current status is: %s", connected_peers.size(), getStatus().c_str());
+ begin_receive_time = simgrid::s4u::Engine::getClock();
+ mailbox_->setReceiver(simgrid::s4u::Actor::self());
+ if (hasFinished()) {
+ sendHandshakeToAllPeers();
+ } else {
+ leech();
+ }
+ seed();
+ } else {
+ XBT_INFO("Couldn't contact the tracker.");
+ }
+
+ XBT_INFO("Here is my current status: %s", getStatus().c_str());
+}
+
+bool Peer::getPeersFromTracker()
+{
+ simgrid::s4u::MailboxPtr tracker_mailbox = simgrid::s4u::Mailbox::byName(TRACKER_MAILBOX);
+ // Build the task to send to the tracker
+ TrackerQuery* peer_request = new TrackerQuery(id, mailbox_, 0, 0, FILE_SIZE);
+ try {
+ XBT_DEBUG("Sending a peer request to the tracker.");
+ tracker_mailbox->put(peer_request, TRACKER_COMM_SIZE, GET_PEERS_TIMEOUT);
+ } catch (xbt_ex& e) {
+ if (e.category == timeout_error) {
+ XBT_DEBUG("Timeout expired when requesting peers to tracker");
+ delete peer_request;
+ return false;
+ }
+ }
+
+ try {
+ TrackerAnswer* answer = static_cast<TrackerAnswer*>(mailbox_->get(GET_PEERS_TIMEOUT));
+ // Add the peers the tracker gave us to our peer list.
+ for (auto peer_id : *answer->getPeers())
+ if (id != peer_id)
+ connected_peers[peer_id] = new Connection(peer_id);
+ delete answer;
+ } catch (xbt_ex& e) {
+ if (e.category == timeout_error) {
+ XBT_DEBUG("Timeout expired when requesting peers to tracker");
+ return false;
+ }
+ }
+ return true;
+}
+
+void Peer::sendHandshakeToAllPeers()
+{
+ for (auto kv : connected_peers) {
+ Connection* remote_peer = kv.second;
+ Message* handshake = new Message(MESSAGE_HANDSHAKE, id, mailbox_);
+ remote_peer->mailbox_->put_init(handshake, MESSAGE_HANDSHAKE_SIZE)->detach();
+ XBT_DEBUG("Sending a HANDSHAKE to %d", remote_peer->id);
+ }
+}
+
+void Peer::sendHandshake(simgrid::s4u::MailboxPtr mailbox)
+{
+ XBT_DEBUG("Sending a HANDSHAKE to %s", mailbox->getName());
+ mailbox->put_init(new Message(MESSAGE_HANDSHAKE, id, mailbox_), MESSAGE_HANDSHAKE_SIZE)->detach();
+}
+
+void Peer::sendBitfield(simgrid::s4u::MailboxPtr mailbox)
+{
+ XBT_DEBUG("Sending a BITFIELD to %s", mailbox->getName());
+ mailbox
+ ->put_init(new Message(MESSAGE_BITFIELD, id, bitfield_, mailbox_),
+ MESSAGE_BITFIELD_SIZE + BITS_TO_BYTES(FILE_PIECES))
+ ->detach();
+}
+
+void Peer::sendInterested(simgrid::s4u::MailboxPtr mailbox)
+{
+ XBT_DEBUG("Sending INTERESTED to %s", mailbox->getName());
+ mailbox->put_init(new Message(MESSAGE_INTERESTED, id, bitfield_, mailbox_), MESSAGE_INTERESTED_SIZE)->detach();
+}
+
+void Peer::sendNotInterested(simgrid::s4u::MailboxPtr mailbox)
+{
+ XBT_DEBUG("Sending NOTINTERESTED to %s", mailbox->getName());
+ mailbox->put_init(new Message(MESSAGE_NOTINTERESTED, id, bitfield_, mailbox_), MESSAGE_NOTINTERESTED_SIZE)->detach();
+}
+
+void Peer::sendChoked(simgrid::s4u::MailboxPtr mailbox)
+{
+ XBT_DEBUG("Sending CHOKE to %s", mailbox->getName());
+ mailbox->put_init(new Message(MESSAGE_CHOKE, id, mailbox_), MESSAGE_CHOKE_SIZE)->detach();
+}
+
+/** Send a "unchoked" message to a peer */
+void Peer::sendUnchoked(simgrid::s4u::MailboxPtr mailbox)
+{
+ XBT_DEBUG("Sending UNCHOKE to %s", mailbox->getName());
+ mailbox->put_init(new Message(MESSAGE_UNCHOKE, id, mailbox_), MESSAGE_UNCHOKE_SIZE)->detach();
+}
+
+void Peer::sendPiece(simgrid::s4u::MailboxPtr mailbox, unsigned int piece, int block_index, int block_length)
+{
+ xbt_assert(!hasNotPiece(piece), "Tried to send a unavailable piece.");
+ XBT_DEBUG("Sending the PIECE %u (%d,%d) to %s", piece, block_index, block_length, mailbox->getName());
+ mailbox->put_init(new Message(MESSAGE_PIECE, id, mailbox_, piece, block_index, block_length), BLOCK_SIZE)->detach();
+}
+
+void Peer::sendRequest(simgrid::s4u::MailboxPtr mailbox, unsigned int piece, int block_index, int block_length)
+{
+ XBT_DEBUG("Sending a REQUEST to %s for piece %u (%d,%d)", mailbox->getName(), piece, block_index, block_length);
+ mailbox->put_init(new Message(MESSAGE_REQUEST, id, mailbox_, piece, block_index, block_length), MESSAGE_REQUEST_SIZE)
+ ->detach();
+}
+
+void Peer::sendHaveToAllPeers(unsigned int piece)
+{
+ XBT_DEBUG("Sending HAVE message to all my peers");
+ for (auto kv : connected_peers) {
+ Connection* remote_peer = kv.second;
+ remote_peer->mailbox_->put_init(new Message(MESSAGE_HAVE, id, mailbox_, piece), MESSAGE_HAVE_SIZE)->detach();
+ }
+}
+
+void Peer::sendRequestTo(Connection* remote_peer, unsigned int piece)
+{
+ remote_peer->current_piece = piece;
+ xbt_assert(remote_peer->hasPiece(piece));
+ int block_index = getFirstMissingBlockFrom(piece);
+ if (block_index != -1) {
+ int block_length = MIN(BLOCKS_REQUESTED, PIECES_BLOCKS - block_index);
+ sendRequest(remote_peer->mailbox_, piece, block_index, block_length);
+ }
+}
+
+std::string Peer::getStatus()
+{
+ std::string res = std::string("");
+ for (int i = FILE_PIECES - 1; i >= 0; i--)
+ res = std::string((bitfield_ & (1U << i)) ? "1" : "0") + res;
+ return res;
+}
+
+bool Peer::hasFinished()
+{
+ return bitfield_ == (1U << FILE_PIECES) - 1U;
+}
+
+/** Indicates if the remote peer has a piece not stored by the local peer */
+bool Peer::isInterestedBy(Connection* remote_peer)
+{
+ return remote_peer->bitfield & (bitfield_ ^ ((1 << FILE_PIECES) - 1));
+}
+
+void Peer::updatePiecesCountFromBitfield(unsigned int bitfield)
+{
+ for (unsigned int i = 0; i < FILE_PIECES; i++)
+ if (bitfield & (1U << i))
+ pieces_count[i]++;
+}
+
+unsigned int Peer::countPieces(unsigned int bitfield)
+{
+ unsigned int count = 0U;
+ unsigned int n = bitfield;
+ while (n) {
+ count += n & 1U;
+ n >>= 1U;
+ }
+ return count;
+}
+
+int Peer::nbInterestedPeers()
+{
+ int nb = 0;
+ for (auto kv : connected_peers)
+ if (kv.second->interested)
+ nb++;
+ return nb;
+}
+
+void Peer::leech()
+{
+ double next_choked_update = simgrid::s4u::Engine::getClock() + UPDATE_CHOKED_INTERVAL;
+ XBT_DEBUG("Start downloading.");
+
+ /* Send a "handshake" message to all the peers it got (since it couldn't have gotten more than 50 peers) */
+ sendHandshakeToAllPeers();
+ XBT_DEBUG("Starting main leech loop listening on mailbox: %s", mailbox_->getName());
+
+ void* data = nullptr;
+ while (simgrid::s4u::Engine::getClock() < deadline && countPieces(bitfield_) < FILE_PIECES) {
+ if (comm_received == nullptr) {
+ comm_received = mailbox_->get_async(&data);
+ }
+ if (comm_received->test()) {
+ message = static_cast<Message*>(data);
+ handleMessage();
+ delete message;
+ comm_received = nullptr;
+ } else {
+ // We don't execute the choke algorithm if we don't already have a piece
+ if (simgrid::s4u::Engine::getClock() >= next_choked_update && countPieces(bitfield_) > 0) {
+ updateChokedPeers();
+ next_choked_update += UPDATE_CHOKED_INTERVAL;
+ } else {
+ simgrid::s4u::this_actor::sleep_for(SLEEP_DURATION);
+ }
+ }
+ }
+ if (hasFinished())
+ XBT_DEBUG("%d becomes a seeder", id);
+}
+
+void Peer::seed()
+{
+ double next_choked_update = simgrid::s4u::Engine::getClock() + UPDATE_CHOKED_INTERVAL;
+ XBT_DEBUG("Start seeding.");
+ // start the main seed loop
+ void* data = nullptr;
+ while (simgrid::s4u::Engine::getClock() < deadline) {
+ if (comm_received == nullptr) {
+ comm_received = mailbox_->get_async(&data);
+ }
+ if (comm_received->test()) {
+ message = static_cast<Message*>(data);
+ handleMessage();
+ delete message;
+ comm_received = nullptr;
+ } else {
+ if (simgrid::s4u::Engine::getClock() >= next_choked_update) {
+ updateChokedPeers();
+ // TODO: Change the choked peer algorithm when seeding.
+ next_choked_update += UPDATE_CHOKED_INTERVAL;
+ } else {
+ simgrid::s4u::this_actor::sleep_for(SLEEP_DURATION);
+ }
+ }
+ }
+}
+
+void Peer::updateActivePeersSet(Connection* remote_peer)
+{
+ if (remote_peer->interested && not remote_peer->choked_upload) {
+ // add in the active peers set
+ active_peers.insert(remote_peer);
+ } else if (active_peers.find(remote_peer) != active_peers.end()) {
+ active_peers.erase(remote_peer);
+ }
+}
+
+void Peer::handleMessage()
+{
+ const char* type_names[10] = {"HANDSHAKE", "CHOKE", "UNCHOKE", "INTERESTED", "NOTINTERESTED",
+ "HAVE", "BITFIELD", "REQUEST", "PIECE", "CANCEL"};
+
+ XBT_DEBUG("Received a %s message from %s", type_names[message->type], message->return_mailbox->getName());
+
+ auto known_peer = connected_peers.find(message->peer_id);
+ Connection* remote_peer = (known_peer == connected_peers.end()) ? nullptr : known_peer->second;
+ switch (message->type) {
+ case MESSAGE_HANDSHAKE:
+ // Check if the peer is in our connection list.
+ if (remote_peer == nullptr) {
+ XBT_DEBUG("This peer %d was unknown, answer to its handshake", message->peer_id);
+ connected_peers[message->peer_id] = new Connection(message->peer_id);
+ sendHandshake(message->return_mailbox);
+ }
+ // Send our bitfield to the peer
+ sendBitfield(message->return_mailbox);
+ break;
+ case MESSAGE_BITFIELD:
+ // Update the pieces list
+ updatePiecesCountFromBitfield(message->bitfield);
+ // Store the bitfield
+ remote_peer->bitfield = message->bitfield;
+ xbt_assert(!remote_peer->am_interested, "Should not be interested at first");
+ if (isInterestedBy(remote_peer)) {
+ remote_peer->am_interested = true;
+ sendInterested(message->return_mailbox);
+ }
+ break;
+ case MESSAGE_INTERESTED:
+ xbt_assert((remote_peer != nullptr), "The impossible did happened: A not-in-our-list peer sent us a message.");
+ // Update the interested state of the peer.
+ remote_peer->interested = true;
+ updateActivePeersSet(remote_peer);
+ break;
+ case MESSAGE_NOTINTERESTED:
+ xbt_assert((remote_peer != nullptr), "The impossible did happened: A not-in-our-list peer sent us a message.");
+ remote_peer->interested = false;
+ updateActivePeersSet(remote_peer);
+ break;
+ case MESSAGE_UNCHOKE:
+ xbt_assert((remote_peer != nullptr), "The impossible did happened: A not-in-our-list peer sent us a message.");
+ xbt_assert(remote_peer->choked_download);
+ remote_peer->choked_download = false;
+ // Send requests to the peer, since it has unchoked us
+ // if (remote_peer->am_interested)
+ requestNewPieceTo(remote_peer);
+ break;
+ case MESSAGE_CHOKE:
+ xbt_assert((remote_peer != nullptr), "The impossible did happened: A not-in-our-list peer sent us a message.");
+ xbt_assert(not remote_peer->choked_download);
+ remote_peer->choked_download = true;
+ if (remote_peer->current_piece != -1)
+ removeCurrentPiece(remote_peer, remote_peer->current_piece);
+ break;
+ case MESSAGE_HAVE:
+ XBT_DEBUG("\t for piece %d", message->piece);
+ xbt_assert((message->piece >= 0 && static_cast<unsigned int>(message->piece) < FILE_PIECES),
+ "Wrong HAVE message received");
+ remote_peer->bitfield = remote_peer->bitfield | (1U << static_cast<unsigned int>(message->piece));
+ pieces_count[message->piece]++;
+ // If the piece is in our pieces, we tell the peer that we are interested.
+ if (not remote_peer->am_interested && hasNotPiece(message->piece)) {
+ remote_peer->am_interested = true;
+ sendInterested(message->return_mailbox);
+ if (not remote_peer->choked_download)
+ requestNewPieceTo(remote_peer);
+ }
+ break;
+ case MESSAGE_REQUEST:
+ xbt_assert(remote_peer->interested);
+ xbt_assert((message->piece >= 0 && static_cast<unsigned int>(message->piece) < FILE_PIECES),
+ "Wrong HAVE message received");
+ if (not remote_peer->choked_upload) {
+ XBT_DEBUG("\t for piece %d (%d,%d)", message->piece, message->block_index,
+ message->block_index + message->block_length);
+ if (not hasNotPiece(message->piece)) {
+ sendPiece(message->return_mailbox, message->piece, message->block_index, message->block_length);
+ }
+ } else {
+ XBT_DEBUG("\t for piece %d but he is choked.", message->peer_id);
+ }
+ break;
+ case MESSAGE_PIECE:
+ XBT_DEBUG(" \t for piece %d (%d,%d)", message->piece, message->block_index,
+ message->block_index + message->block_length);
+ xbt_assert(not remote_peer->choked_download);
+ xbt_assert(remote_peer->am_interested || ENABLE_END_GAME_MODE,
+ "Can't received a piece if I'm not interested wihtout end-game mode!"
+ "piece (%d) bitfield (%u) remote bitfield (%u)",
+ message->piece, bitfield_, remote_peer->bitfield);
+ xbt_assert(not remote_peer->choked_download, "Can't received a piece if I'm choked !");
+ xbt_assert((message->piece >= 0 && static_cast<unsigned int>(message->piece) < FILE_PIECES),
+ "Wrong piece received");
+ // TODO: Execute a computation.
+ if (hasNotPiece(static_cast<unsigned int>(message->piece))) {
+ updateBitfieldBlocks(message->piece, message->block_index, message->block_length);
+ if (hasCompletedPiece(static_cast<unsigned int>(message->piece))) {
+ // Removing the piece from our piece list
+ removeCurrentPiece(remote_peer, message->piece);
+ // Setting the fact that we have the piece
+ bitfield_ = bitfield_ | (1U << static_cast<unsigned int>(message->piece));
+ XBT_DEBUG("My status is now %s", getStatus().c_str());
+ // Sending the information to all the peers we are connected to
+ sendHaveToAllPeers(message->piece);
+ // sending UNINTERESTED to peers that do not have what we want.
+ updateInterestedAfterReceive();
+ } else { // piece not completed
+ sendRequestTo(remote_peer, message->piece); // ask for the next block
+ }
+ } else {
+ XBT_DEBUG("However, we already have it");
+ xbt_assert(ENABLE_END_GAME_MODE, "Should not happen because we don't use end game mode !");
+ requestNewPieceTo(remote_peer);
+ }
+ break;
+ case MESSAGE_CANCEL:
+ break;
+ default:
+ THROW_IMPOSSIBLE;
+ }
+ // Update the peer speed.
+ if (remote_peer) {
+ remote_peer->addSpeedValue(1.0 / (simgrid::s4u::Engine::getClock() - begin_receive_time));
+ }
+ begin_receive_time = simgrid::s4u::Engine::getClock();
+}
+
+/** Selects the appropriate piece to download and requests it to the remote_peer */
+void Peer::requestNewPieceTo(Connection* remote_peer)
+{
+ int piece = selectPieceToDownload(remote_peer);
+ if (piece != -1) {
+ current_pieces |= (1U << (unsigned int)piece);
+ sendRequestTo(remote_peer, piece);
+ }
+}
+
+void Peer::removeCurrentPiece(Connection* remote_peer, unsigned int current_piece)
+{
+ current_pieces &= ~(1U << current_piece);
+ remote_peer->current_piece = -1;
+}
+
+/** @brief Return the piece to be downloaded
+ * There are two cases (as described in "Bittorrent Architecture Protocol", Ryan Toole :
+ * If a piece is partially downloaded, this piece will be selected prioritarily
+ * If the peer has strictly less than 4 pieces, he chooses a piece at random.
+ * If the peer has more than pieces, he downloads the pieces that are the less replicated (rarest policy).
+ * If all pieces have been downloaded or requested, we select a random requested piece (endgame mode).
+ * @param remote_peer: information about the connection
+ * @return the piece to download if possible. -1 otherwise
+ */
+int Peer::selectPieceToDownload(Connection* remote_peer)
+{
+ int piece = partiallyDownloadedPiece(remote_peer);
+ // strict priority policy
+ if (piece != -1)
+ return piece;
+
+ // end game mode
+ if (countPieces(current_pieces) >= (FILE_PIECES - countPieces(bitfield_)) && isInterestedBy(remote_peer)) {
+#if ENABLE_END_GAME_MODE == 0
+ return -1;
+#endif
+ int nb_interesting_pieces = 0;
+ // compute the number of interesting pieces
+ for (unsigned int i = 0; i < FILE_PIECES; i++)
+ if (hasNotPiece(i) && remote_peer->hasPiece(i))
+ nb_interesting_pieces++;
+
+ xbt_assert(nb_interesting_pieces != 0);
+ // get a random interesting piece
+ int random_piece_index = RngStream_RandInt(stream, 0, nb_interesting_pieces - 1);
+ int current_index = 0;
+ for (unsigned int i = 0; i < FILE_PIECES; i++) {
+ if (hasNotPiece(i) && remote_peer->hasPiece(i)) {
+ if (random_piece_index == current_index) {
+ piece = i;
+ break;
+ }
+ current_index++;
+ }
+ }
+ xbt_assert(piece != -1);
+ return piece;
+ }
+ // Random first policy
+ if (countPieces(bitfield_) < 4 && isInterestedByFree(remote_peer)) {
+ int nb_interesting_pieces = 0;
+ // compute the number of interesting pieces
+ for (unsigned int i = 0; i < FILE_PIECES; i++)
+ if (hasNotPiece(i) && remote_peer->hasPiece(i) && isNotDownloadingPiece(i))
+ nb_interesting_pieces++;
+ xbt_assert(nb_interesting_pieces != 0);
+ // get a random interesting piece
+ int random_piece_index = RngStream_RandInt(stream, 0, nb_interesting_pieces - 1);
+ int current_index = 0;
+ for (unsigned int i = 0; i < FILE_PIECES; i++) {
+ if (hasNotPiece(i) && remote_peer->hasPiece(i) && isNotDownloadingPiece(i)) {
+ if (random_piece_index == current_index) {
+ piece = i;
+ break;
+ }
+ current_index++;
+ }
+ }
+ xbt_assert(piece != -1);
+ return piece;
+ } else { // Rarest first policy
+ short min = SHRT_MAX;
+ int nb_min_pieces = 0;
+ int current_index = 0;
+ // compute the smallest number of copies of available pieces
+ for (unsigned int i = 0; i < FILE_PIECES; i++) {
+ if (pieces_count[i] < min)
+ if (hasNotPiece(i) && remote_peer->hasPiece(i) && isNotDownloadingPiece(i))
+ min = pieces_count[i];
+ }
+
+ xbt_assert(min != SHRT_MAX || not isInterestedByFree(remote_peer));
+ // compute the number of rarest pieces
+ for (unsigned int i = 0; i < FILE_PIECES; i++)
+ if (pieces_count[i] == min && hasNotPiece(i) && remote_peer->hasPiece(i) && isNotDownloadingPiece(i))
+ nb_min_pieces++;
+
+ xbt_assert(nb_min_pieces != 0 || not isInterestedByFree(remote_peer));
+ // get a random rarest piece
+ int random_rarest_index = RngStream_RandInt(stream, 0, nb_min_pieces - 1);
+ for (unsigned int i = 0; i < FILE_PIECES; i++)
+ if (pieces_count[i] == min && hasNotPiece(i) && remote_peer->hasPiece(i) && isNotDownloadingPiece(i)) {
+ if (random_rarest_index == current_index) {
+ piece = i;
+ break;
+ }
+ current_index++;
+ }
+
+ xbt_assert(piece != -1 || not isInterestedByFree(remote_peer));
+ return piece;
+ }
+}
+
+void Peer::updateChokedPeers()
+{
+ if (nbInterestedPeers() == 0)
+ return;
+ XBT_DEBUG("(%d) update_choked peers %zu active peers", id, active_peers.size());
+ // update the current round
+ round_ = (round_ + 1) % 3;
+ Connection* chosen_peer = nullptr;
+ // select first active peer and remove it from the set
+ Connection* choked_peer = *(active_peers.begin());
+ active_peers.erase(choked_peer);
+
+ /**If we are currently seeding, we unchoke the peer which has been unchoked the last time.*/
+ if (hasFinished()) {
+ Connection* remote_peer;
+ double unchoke_time = simgrid::s4u::Engine::getClock() + 1;
+ for (auto kv : connected_peers) {
+ remote_peer = kv.second;
+ if (remote_peer->last_unchoke < unchoke_time && remote_peer->interested && remote_peer->choked_upload) {
+ unchoke_time = remote_peer->last_unchoke;
+ chosen_peer = remote_peer;
+ }
+ }
+ } else {
+ // Random optimistic unchoking
+ if (round_ == 0) {
+ int j = 0;
+ do {
+ // We choose a random peer to unchoke.
+ std::unordered_map<int, Connection*>::iterator chosen_peer_it = connected_peers.begin();
+ std::advance(chosen_peer_it, RngStream_RandInt(stream, 0, connected_peers.size() - 1));
+ chosen_peer = chosen_peer_it->second;
+ if (chosen_peer == nullptr)
+ THROWF(unknown_error, 0, "A peer should have be selected at this point");
+ else if (not chosen_peer->interested || not chosen_peer->choked_upload)
+ chosen_peer = nullptr;
+ else
+ XBT_DEBUG("Nothing to do, keep going");
+ j++;
+ } while (chosen_peer == nullptr && j < MAXIMUM_PEERS);
+ } else {
+ // Use the "fastest download" policy.
+ double fastest_speed = 0.0;
+ for (auto kv : connected_peers) {
+ Connection* remote_peer = kv.second;
+ if (remote_peer->peer_speed > fastest_speed && remote_peer->choked_upload && remote_peer->interested) {
+ chosen_peer = remote_peer;
+ fastest_speed = remote_peer->peer_speed;
+ }
+ }
+ }
+ }
+
+ if (chosen_peer != nullptr)
+ XBT_DEBUG("(%d) update_choked peers unchoked (%d) ; int (%d) ; choked (%d) ", id, chosen_peer->id,
+ chosen_peer->interested, chosen_peer->choked_upload);
+
+ if (choked_peer != chosen_peer) {
+ if (choked_peer != nullptr) {
+ xbt_assert((!choked_peer->choked_upload), "Tries to choked a choked peer");
+ choked_peer->choked_upload = true;
+ updateActivePeersSet(choked_peer);
+ XBT_DEBUG("(%d) Sending a CHOKE to %d", id, choked_peer->id);
+ sendChoked(choked_peer->mailbox_);
+ }
+ if (chosen_peer != nullptr) {
+ xbt_assert((chosen_peer->choked_upload), "Tries to unchoked an unchoked peer");
+ chosen_peer->choked_upload = false;
+ active_peers.insert(chosen_peer);
+ chosen_peer->last_unchoke = simgrid::s4u::Engine::getClock();
+ XBT_DEBUG("(%d) Sending a UNCHOKE to %d", id, chosen_peer->id);
+ updateActivePeersSet(chosen_peer);
+ sendUnchoked(chosen_peer->mailbox_);
+ }
+ }
+}
+
+/** @brief Update "interested" state of peers: send "not interested" to peers that don't have any more pieces we want.*/
+void Peer::updateInterestedAfterReceive()
+{
+ for (auto kv : connected_peers) {
+ Connection* remote_peer = kv.second;
+ if (remote_peer->am_interested) {
+ bool interested = false;
+ // Check if the peer still has a piece we want.
+ for (unsigned int i = 0; i < FILE_PIECES; i++)
+ if (hasNotPiece(i) && remote_peer->hasPiece(i)) {
+ interested = true;
+ break;
+ }
+
+ if (not interested) { // no more piece to download from connection
+ remote_peer->am_interested = false;
+ sendNotInterested(remote_peer->mailbox_);
+ }
+ }
+ }
+}
+
+void Peer::updateBitfieldBlocks(int piece, int block_index, int block_length)
+{
+ xbt_assert((piece >= 0 && static_cast<unsigned int>(piece) <= FILE_PIECES), "Wrong piece.");
+ xbt_assert((block_index >= 0 && static_cast<unsigned int>(block_index) <= PIECES_BLOCKS), "Wrong block : %d.",
+ block_index);
+ for (int i = block_index; i < (block_index + block_length); i++)
+ bitfield_blocks |= (1ULL << static_cast<unsigned int>(piece * PIECES_BLOCKS + i));
+}
+
+bool Peer::hasCompletedPiece(unsigned int piece)
+{
+ for (unsigned int i = 0; i < PIECES_BLOCKS; i++)
+ if (!(bitfield_blocks & 1ULL << (piece * PIECES_BLOCKS + i)))
+ return false;
+ return true;
+}
+
+int Peer::getFirstMissingBlockFrom(int piece)
+{
+ for (unsigned int i = 0; i < PIECES_BLOCKS; i++)
+ if (!(bitfield_blocks & 1ULL << (piece * PIECES_BLOCKS + i)))
+ return i;
+ return -1;
+}
+
+bool Peer::isInterestedByFree(Connection* remote_peer)
+{
+ for (unsigned int i = 0; i < FILE_PIECES; i++)
+ if (hasNotPiece(i) && remote_peer->hasPiece(i) && isNotDownloadingPiece(i))
+ return true;
+ return false;
+}
+
+/** Returns a piece that is partially downloaded and stored by the remote peer if any -1 otherwise. */
+int Peer::partiallyDownloadedPiece(Connection* remote_peer)
+{
+ for (unsigned int i = 0; i < FILE_PIECES; i++)
+ if (hasNotPiece(i) && remote_peer->hasPiece(i) && isNotDownloadingPiece(i) && getFirstMissingBlockFrom(i) > 0)
+ return i;
+ return -1;
+}
--- /dev/null
+/* Copyright (c) 2012-2017. The SimGrid Team.
+ * All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#ifndef BITTORRENT_PEER_HPP
+#define BITTORRENT_PEER_HPP
+#include "s4u_bittorrent.hpp"
+#include <set>
+#include <unordered_map>
+
+class Connection {
+public:
+ int id; // Peer id
+ simgrid::s4u::MailboxPtr mailbox_;
+ unsigned int bitfield = 0U; // Fields
+ // int messages_count;
+ double peer_speed = 0;
+ double last_unchoke = 0;
+ int current_piece = -1;
+ bool am_interested = false; // Indicates if we are interested in something the peer has
+ bool interested = false; // Indicates if the peer is interested in one of our pieces
+ bool choked_upload = true; // Indicates if the peer is choked for the current peer
+ bool choked_download = true; // Indicates if the peer has choked the current peer
+
+ Connection(int id) : id(id), mailbox_(simgrid::s4u::Mailbox::byName(std::to_string(id))){};
+ ~Connection() = default;
+ void addSpeedValue(double speed) { peer_speed = peer_speed * 0.6 + speed * 0.4; }
+ bool hasPiece(unsigned int piece) { return bitfield & 1U << piece; }
+};
+
+class Peer {
+ int id;
+ double deadline;
+ RngStream stream;
+ simgrid::s4u::MailboxPtr mailbox_;
+ std::set<Connection*> active_peers; // active peers list
+
+ unsigned int bitfield_ = 0; // list of pieces the peer has.
+ unsigned long long bitfield_blocks = 0; // list of blocks the peer has.
+ short* pieces_count = nullptr; // number of peers that have each piece.
+ unsigned int current_pieces = 0; // current pieces the peer is downloading
+ double begin_receive_time = 0; // time when the receiving communication has begun, useful for calculating host speed.
+ int round_ = 0; // current round for the chocking algorithm.
+
+ std::unordered_map<int, Connection*> connected_peers;
+ simgrid::s4u::CommPtr comm_received = nullptr; // current comm
+ Message* message = nullptr; // current message being received
+public:
+ explicit Peer(std::vector<std::string> args);
+ ~Peer();
+ void operator()();
+
+ std::string getStatus();
+ bool hasFinished();
+ int nbInterestedPeers();
+ bool isInterestedBy(Connection* remote_peer);
+ bool isInterestedByFree(Connection* remote_peer);
+ void updateActivePeersSet(Connection* remote_peer);
+ void updateInterestedAfterReceive();
+ void updateChokedPeers();
+
+ bool hasNotPiece(unsigned int piece) { return !(bitfield_ & 1U << piece); }
+ bool hasCompletedPiece(unsigned int piece);
+ unsigned int countPieces(unsigned int bitfield);
+ /** Check that a piece is not currently being download by the peer. */
+ bool isNotDownloadingPiece(unsigned int piece) { return !(current_pieces & 1U << piece); }
+ int partiallyDownloadedPiece(Connection* remote_peer);
+ void updatePiecesCountFromBitfield(unsigned int bitfield);
+ void removeCurrentPiece(Connection* remote_peer, unsigned int current_piece);
+ void updateBitfieldBlocks(int piece, int block_index, int block_length);
+ int getFirstMissingBlockFrom(int piece);
+ int selectPieceToDownload(Connection* remote_peer);
+ void requestNewPieceTo(Connection* remote_peer);
+
+ bool getPeersFromTracker();
+ void sendHandshake(simgrid::s4u::MailboxPtr mailbox);
+ void sendBitfield(simgrid::s4u::MailboxPtr mailbox);
+ void sendPiece(simgrid::s4u::MailboxPtr mailbox, unsigned int piece, int block_index, int block_length);
+ void sendInterested(simgrid::s4u::MailboxPtr mailbox);
+ void sendChoked(simgrid::s4u::MailboxPtr mailbox);
+ void sendUnchoked(simgrid::s4u::MailboxPtr mailbox);
+ void sendNotInterested(simgrid::s4u::MailboxPtr mailbox);
+ void sendHandshakeToAllPeers();
+ void sendRequest(simgrid::s4u::MailboxPtr mailbox, unsigned int piece, int block_index, int block_length);
+ void sendHaveToAllPeers(unsigned int piece);
+ void sendRequestTo(Connection* remote_peer, unsigned int piece);
+
+ void handleMessage();
+ void leech();
+ void seed();
+};
+
+#endif /* BITTORRENT_PEER_HPP */
--- /dev/null
+/* Copyright (c) 2012-2017. The SimGrid Team.
+ * All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#include "s4u_tracker.hpp"
+#include <xbt/RngStream.h>
+
+XBT_LOG_NEW_DEFAULT_CATEGORY(s4u_bt_tracker, "Messages specific for the tracker");
+
+Tracker::Tracker(std::vector<std::string> args)
+{
+ // Checking arguments
+ xbt_assert(args.size() == 2, "Wrong number of arguments for the tracker.");
+ // Retrieving end time
+ try {
+ deadline = std::stod(args[1]);
+ } catch (std::invalid_argument& ia) {
+ throw std::invalid_argument(std::string("Invalid deadline:") + args[1].c_str());
+ }
+ xbt_assert(deadline > 0, "Wrong deadline supplied");
+
+ stream = simgrid::s4u::this_actor::getHost()->extension<HostBittorrent>()->getStream();
+
+ mailbox = simgrid::s4u::Mailbox::byName(TRACKER_MAILBOX);
+
+ XBT_INFO("Tracker launched.");
+}
+
+void Tracker::operator()()
+{
+ simgrid::s4u::CommPtr comm = nullptr;
+ while (simgrid::s4u::Engine::getClock() < deadline) {
+ void* received;
+ if (comm == nullptr)
+ comm = mailbox->get_async(&received);
+ if (comm->test()) {
+ // Retrieve the data sent by the peer.
+ TrackerQuery* tq = static_cast<TrackerQuery*>(received);
+
+ // Add the peer to our peer list, if not already known.
+ if (known_peers.find(tq->getPeerId()) == known_peers.end()) {
+ known_peers.insert(tq->getPeerId());
+ }
+
+ // Sending back peers to the requesting peer
+ TrackerAnswer* ta = new TrackerAnswer(TRACKER_QUERY_INTERVAL);
+ std::set<int>::iterator next_peer;
+ int nb_known_peers = known_peers.size();
+ int max_tries = MIN(MAXIMUM_PEERS, nb_known_peers);
+ int tried = 0;
+ while (tried < max_tries) {
+ do {
+ next_peer = known_peers.begin();
+ std::advance(next_peer, RngStream_RandInt(stream, 0, nb_known_peers - 1));
+ } while (ta->getPeers()->find(*next_peer) != ta->getPeers()->end());
+ ta->addPeer(*next_peer);
+ tried++;
+ }
+ tq->getReturnMailbox()->put_init(ta, TRACKER_COMM_SIZE)->detach();
+
+ delete tq;
+ comm = nullptr;
+ } else {
+ simgrid::s4u::this_actor::sleep_for(1);
+ }
+ }
+ // TODO See if some cleanup is needed
+ XBT_INFO("Tracker is leaving");
+}
--- /dev/null
+/* Copyright (c) 2012-2014, 2017. The SimGrid Team.
+ * All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#ifndef BITTORRENT_TRACKER_HPP_
+#define BITTORRENT_TRACKER_HPP_
+
+#include "s4u_bittorrent.hpp"
+#include <set>
+
+class TrackerQuery {
+ int peer_id; // peer id
+ simgrid::s4u::MailboxPtr return_mailbox;
+ int uploaded; // how much the peer has already uploaded
+ int downloaded; // how much the peer has downloaded
+ int left; // how much the peer has left
+public:
+ explicit TrackerQuery(int peer_id, simgrid::s4u::MailboxPtr return_mailbox, int uploaded, int downloaded, int left)
+ : peer_id(peer_id), return_mailbox(return_mailbox), uploaded(uploaded), downloaded(downloaded), left(left){};
+ ~TrackerQuery() = default;
+ int getPeerId() { return peer_id; }
+ simgrid::s4u::MailboxPtr getReturnMailbox() { return return_mailbox; }
+};
+
+class TrackerAnswer {
+ int interval; // how often the peer should contact the tracker (unused for now)
+ std::set<int>* peers; // the peer list the peer has asked for.
+public:
+ explicit TrackerAnswer(int interval) : interval(interval) { peers = new std::set<int>; }
+ ~TrackerAnswer() { delete peers; };
+ void addPeer(int peer) { peers->insert(peer); }
+ std::set<int>* getPeers() { return peers; }
+};
+
+class Tracker {
+ double deadline;
+ RngStream stream;
+ simgrid::s4u::MailboxPtr mailbox;
+ std::set<int> known_peers;
+
+public:
+ explicit Tracker(std::vector<std::string> args);
+ void operator()();
+};
+
+#endif /* BITTORRENT_TRACKER_HPP */
> [84.067138] [test/INFO] ------------------- Produce the trace file---------------------------
> [84.067138] [test/INFO] Producing the trace of the run into smalldax.trace
-$ cat ${srcdir:=.}/daxload/smalldax.trace
+$ sort ${srcdir:=.}/daxload/smalldax.trace
> [0.000000] node-0.acme.org compute 0.000000 # root
-> [0.016600] node-1.acme.org compute 42000000000.000000 # 1@task1
-> [0.016600] node-10.acme.org compute 42000000000.000000 # 2@task2
-> [42.033200] node-11.acme.org compute 42000000000.000000 # 3@task1
> [0.000000] node-0.acme.org send node-10.acme.org 1000000.000000 # root_i2_2@task2
+> [0.000000] node-0.acme.org send node-1.acme.org 1000000.000000 # root_i1_1@task1
+> [0.016600] node-10.acme.org compute 42000000000.000000 # 2@task2
> [0.016600] node-10.acme.org recv node-0.acme.org 1000000.000000 # root_i2_2@task2
-> [42.016600] node-1.acme.org send node-11.acme.org 1000000.000000 # 1@task1_o1_3@task1
-> [42.033200] node-11.acme.org recv node-1.acme.org 1000000.000000 # 1@task1_o1_3@task1
+> [0.016600] node-1.acme.org compute 42000000000.000000 # 1@task1
+> [0.016600] node-1.acme.org recv node-0.acme.org 1000000.000000 # root_i1_1@task1
> [42.016600] node-10.acme.org send node-11.acme.org 1000000.000000 # 2@task2_o2_3@task1
+> [42.016600] node-1.acme.org send node-11.acme.org 1000000.000000 # 1@task1_o1_3@task1
+> [42.033200] node-11.acme.org compute 42000000000.000000 # 3@task1
> [42.033200] node-11.acme.org recv node-10.acme.org 1000000.000000 # 2@task2_o2_3@task1
+> [42.033200] node-11.acme.org recv node-1.acme.org 1000000.000000 # 1@task1_o1_3@task1
> [84.033200] node-11.acme.org send node-0.acme.org 4167312.000000 # 3@task1_o3_end
-> [84.067138] node-0.acme.org recv node-11.acme.org 4167312.000000 # 3@task1_o3_end
-> [0.000000] node-0.acme.org send node-1.acme.org 1000000.000000 # root_i1_1@task1
-> [0.016600] node-1.acme.org recv node-0.acme.org 1000000.000000 # root_i1_1@task1
> [84.067138] node-0.acme.org compute 0.000000 # end
+> [84.067138] node-0.acme.org recv node-11.acme.org 4167312.000000 # 3@task1_o3_end
$ cmake -E remove -f ${srcdir:=.}/dax.dot ${srcdir:=.}/daxload/smalldax.trace
-/* Copyright (c) 2004-2015. The SimGrid Team.
- * All rights reserved. */
+/* Copyright (c) 2004-2017. The SimGrid Team. All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
XBT_PUBLIC(msg_comm_t) MSG_task_isend(msg_task_t task, const char *alias);
XBT_PUBLIC(msg_comm_t) MSG_task_isend_bounded(msg_task_t task, const char *alias, double maxrate);
XBT_PUBLIC(msg_comm_t)
-MSG_task_isend_with_matching(msg_task_t task, const char* alias, int (*match_fun)(void*, void*, void*),
- void* match_data);
+XBT_ATTRIB_DEPRECATED_v320(
+ "This function will be removed from SimGrid v3.20. If you really need this function, please speak up quickly.")
+ MSG_task_isend_with_matching(msg_task_t task, const char* alias, int (*match_fun)(void*, void*, void*),
+ void* match_data);
XBT_PUBLIC(void) MSG_task_dsend(msg_task_t task, const char *alias, void_f_pvoid_t cleanup);
XBT_PUBLIC(void) MSG_task_dsend_bounded(msg_task_t task, const char *alias, void_f_pvoid_t cleanup, double maxrate);
/* Used only by the bindings -- unclean pimple, please ignore if you're not writing a binding */
XBT_PUBLIC(smx_context_t) MSG_process_get_smx_ctx(msg_process_t process);
-/* Functions renamed in 3.14 */
-#define MSG_mailbox_get_head(m) MSG_mailbox_front(m)
-
SG_END_DECL()
#ifdef __cplusplus
#include <unordered_map>
#include "xbt/Extendable.hpp"
-#include "xbt/dict.h"
#include "xbt/signal.hpp"
#include "xbt/string.hpp"
#include "xbt/swag.h"
SG_BEGIN_DECL()
-XBT_PUBLIC(xbt_dynar_t) SIMIX_process_get_runnable();
XBT_PUBLIC(smx_actor_t) SIMIX_process_from_PID(aid_t PID);
XBT_PUBLIC(xbt_dynar_t) SIMIX_processes_as_dynar();
-/* Copyright (c) 2007-2010, 2012-2015. The SimGrid Team.
+/* Copyright (c) 2007-2010, 2012-2017. The SimGrid Team.
* All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
return result.get();
}
+XBT_PUBLIC(const std::vector<smx_actor_t>&) process_get_runnable();
XBT_PUBLIC(void) set_maestro(std::function<void()> code);
XBT_PUBLIC(void) create_maestro(std::function<void()> code);
--- /dev/null
+/* Copyright (c) 2015-2017. The SimGrid Team. All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#ifndef XBT_ALGORITHM_HPP
+#define XBT_ALGORITHM_HPP
+
+namespace simgrid {
+namespace xbt {
+
+/** @brief Sorts the elements of the sequence [first, last) according to their color assuming elements can have only
+ * three colors. Since there are only three colors, it is linear and much faster than a classical sort. See for
+ * example http://en.wikipedia.org/wiki/Dutch_national_flag_problem
+ *
+ * \param first forward iterators to the initial position of the sequence to partition.
+ * \param last forward iterators to the final position of the sequence to partition.
+ * \param color the color function that accepts an element in the range as argument, and returns a value of 0, 1, or 2.
+ *
+ * At the end of the call, elements with color 0 are at the beginning of the range, elements with color 2 are at the end
+ * and elements with color 1 are in the middle.
+ */
+template <class ForwardIterator, class ColorFunction>
+void three_way_partition(ForwardIterator first, ForwardIterator last, ColorFunction color)
+{
+ ForwardIterator iter = first;
+ while (iter < last) {
+ int c = color(*iter);
+ if (c == 1) {
+ ++iter;
+ } else if (c == 0) {
+ if (iter != first)
+ std::swap(*iter, *first);
+ ++iter;
+ ++first;
+ } else { // c == 2
+ --last;
+ if (iter != last)
+ std::swap(*iter, *last);
+ }
+ }
+}
+}
+}
+
+#endif
XBT_PUBLIC(int) xbt_dynar_member(xbt_dynar_t const dynar, void *elem);
XBT_PUBLIC(void) xbt_dynar_sort(xbt_dynar_t const dynar, int_f_cpvoid_cpvoid_t compar_fn);
XBT_PUBLIC(xbt_dynar_t) xbt_dynar_sort_strings(xbt_dynar_t dynar);
-XBT_PUBLIC(void) xbt_dynar_three_way_partition(xbt_dynar_t const dynar, int_f_pvoid_t color);
XBT_PUBLIC(int) xbt_dynar_compare(xbt_dynar_t d1, xbt_dynar_t d2, int(*compar)(const void *, const void *));
XBT_PUBLIC(void *) xbt_dynar_to_array (xbt_dynar_t dynar);
/* function_type.h - classical types for pointer to function */
-/* Copyright (c) 2006-2007, 2009-2010, 2012-2015. The SimGrid Team.
- * All rights reserved. */
+/* Copyright (c) 2006-2017. The SimGrid Team. All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
typedef void *(*pvoid_f_pvoid_t) (void *);
typedef void (*void_f_void_t) (void);
-typedef int (*int_f_pvoid_t) (void*);
-
typedef int (*int_f_pvoid_pvoid_t) (void *, void *);
typedef int (*int_f_cpvoid_cpvoid_t) (const void *, const void *);
/* A thread pool. */
-/* Copyright (c) 2007, 2009-2014. The SimGrid Team.
+/* Copyright (c) 2007, 2009-2014, 2016-2017. The SimGrid Team.
* All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
#ifndef XBT_PARMAP_H
#define XBT_PARMAP_H
-#include "xbt/misc.h" /* SG_BEGIN_DECL */
-#include "xbt/function_types.h"
-#include "xbt/dynar.h"
-
-SG_BEGIN_DECL()
-
/** \addtogroup XBT_parmap
* \ingroup XBT_misc
* \brief Parallel map.
*
- * A function is applied to all elements of a dynar in parallel with n worker threads.
- * The worker threads are persistent until the destruction of the parmap.
+ * A function is applied to all elements of a std::vector in parallel with n worker threads. The worker threads are
+ * persistent until the destruction of the parmap.
*
- * If there are more than n elements in the dynar, the worker threads are allowed to fetch themselves remaining work
- * with xbt_parmap_next() and execute it.
+ * If there are more than n elements in the vector, the worker threads are allowed to fetch themselves remaining work
+ * with method next() and execute it.
*
* \{
*/
-/** \brief Parallel map data type (opaque type) */
-typedef struct s_xbt_parmap *xbt_parmap_t;
-
/** \brief Synchronization mode of the worker threads of a parmap. */
typedef enum {
XBT_PARMAP_POSIX, /**< use POSIX synchronization primitives */
XBT_PARMAP_DEFAULT /**< futex if available, posix otherwise */
} e_xbt_parmap_mode_t;
-XBT_PUBLIC(xbt_parmap_t) xbt_parmap_new(unsigned int num_workers, e_xbt_parmap_mode_t mode);
-XBT_PUBLIC(void) xbt_parmap_destroy(xbt_parmap_t parmap);
-XBT_PUBLIC(void) xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data);
-XBT_PUBLIC(void*) xbt_parmap_next(xbt_parmap_t parmap);
-
/** \} */
-SG_END_DECL()
-
#endif
void JavaContextFactory::run_all()
{
- xbt_dynar_t processes = SIMIX_process_get_runnable();
- smx_actor_t process;
- unsigned int cursor;
- xbt_dynar_foreach(processes, cursor, process) {
+ for (smx_actor_t process : simgrid::simix::process_get_runnable()) {
static_cast<JavaContext*>(process->context)->resume();
}
}
#include "simgrid/plugins/energy.h"
#include "simgrid/s4u/Host.hpp"
+#include "simgrid/s4u/Storage.hpp"
#include "jmsg.h"
#include "jmsg_host.h"
int index = 0;
jobjectArray jtable;
- xbt_dict_t dict = MSG_host_get_mounted_storage_list(host);
- int count = xbt_dict_length(dict);
+ std::unordered_map<std::string, msg_storage_t> mounted_storages = host->getMountedStorages();
+ int count = mounted_storages.size();
jclass cls = env->FindClass("org/simgrid/msg/Storage");
jtable = env->NewObjectArray((jsize) count, cls, nullptr);
return nullptr;
}
- xbt_dict_cursor_t cursor=nullptr;
- const char* mount_name;
- const char* storage_name;
-
- xbt_dict_foreach(dict,cursor,mount_name,storage_name) {
- jname = env->NewStringUTF(storage_name);
+ for (auto elm : mounted_storages) {
+ jname = env->NewStringUTF(elm.second->getName());
jstorage = Java_org_simgrid_msg_Storage_getByName(env,cls,jname);
env->SetObjectArrayElement(jtable, index, jstorage);
index++;
}
- xbt_dict_free(&dict);
return jtable;
}
-/* Copyright (c) 2010, 2012-2017. The SimGrid Team.
- * All rights reserved. */
+/* Copyright (c) 2010-2017. The SimGrid Team. All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
#include "src/surf/surf_private.h"
#include <boost/algorithm/string/classification.hpp>
#include <boost/algorithm/string/split.hpp>
-#include <simgrid/host.h>
+#include <simgrid/s4u/Host.hpp>
#include <string>
#include <vector>
}
int console_AS_open(lua_State *L) {
- const char *id;
- const char *mode;
- int type;
-
XBT_DEBUG("Opening AS");
lua_ensure(lua_istable(L, 1), "Bad Arguments to AS_open, Should be a table with named arguments");
lua_pushstring(L, "id");
- type = lua_gettable(L, -2);
+ int type = lua_gettable(L, -2);
lua_ensure(type == LUA_TSTRING, "Attribute 'id' must be specified for any AS and must be a string.");
- id = lua_tostring(L, -1);
+ const char* id = lua_tostring(L, -1);
lua_pop(L, 1);
lua_pushstring(L, "mode");
lua_gettable(L, -2);
- mode = lua_tostring(L, -1);
+ const char* mode = lua_tostring(L, -1);
lua_pop(L, 1);
int mode_int = A_surfxml_AS_routing_None;
}
int console_host_set_property(lua_State *L) {
- const char* name ="";
- const char* prop_id = "";
- const char* prop_value = "";
lua_ensure(lua_istable(L, -1), "Bad Arguments to create link, Should be a table with named arguments");
// get Host id
lua_pushstring(L, "host");
lua_gettable(L, -2);
- name = lua_tostring(L, -1);
+ const char* name = lua_tostring(L, -1);
lua_pop(L, 1);
// get prop Name
lua_pushstring(L, "prop");
lua_gettable(L, -2);
- prop_id = lua_tostring(L, -1);
+ const char* prop_id = lua_tostring(L, -1);
lua_pop(L, 1);
//get args
lua_pushstring(L,"value");
lua_gettable(L, -2);
- prop_value = lua_tostring(L,-1);
+ const char* prop_value = lua_tostring(L, -1);
lua_pop(L, 1);
sg_host_t host = sg_host_by_name(name);
lua_ensure(host, "no host '%s' found",name);
- xbt_dict_t props = sg_host_get_properties(host);
- xbt_dict_set(props,prop_id,xbt_strdup(prop_value),nullptr);
+ host->setProperty(prop_id, prop_value);
return 0;
}
XBT_PUBLIC(int) TRACE_end ();
XBT_PUBLIC(void) TRACE_global_init();
XBT_PUBLIC(void) TRACE_help(int detailed);
-XBT_PUBLIC(void) TRACE_surf_resource_utilization_alloc();
-XBT_PUBLIC(void) TRACE_surf_resource_utilization_release();
SG_END_DECL()
/* user-visible parameters */
extern XBT_PRIVATE double sg_tcp_gamma;
-extern XBT_PRIVATE double sg_sender_gap;
extern XBT_PRIVATE double sg_latency_factor;
extern XBT_PRIVATE double sg_bandwidth_factor;
extern XBT_PRIVATE double sg_weight_S_parameter;
/********** Tracing **********/
/* from surf_instr.c */
void TRACE_surf_action(surf_action_t surf_action, const char *category);
-void TRACE_surf_alloc();
-void TRACE_surf_release();
/* instr_routing.c */
void instr_routing_define_callbacks ();
--- /dev/null
+/* A thread pool (C++ version). */
+
+/* Copyright (c) 2004-2017 The SimGrid Team.
+ * All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#ifndef XBT_PARMAP_HPP
+#define XBT_PARMAP_HPP
+
+#include "src/internal_config.h" // HAVE_FUTEX_H
+#include "src/kernel/context/Context.hpp"
+#include <atomic>
+#include <boost/optional.hpp>
+#include <simgrid/simix.h>
+#include <vector>
+#include <xbt/log.h>
+#include <xbt/parmap.h>
+#include <xbt/xbt_os_thread.h>
+
+#if HAVE_FUTEX_H
+#include <limits>
+#include <linux/futex.h>
+#include <sys/syscall.h>
+#endif
+
+XBT_LOG_EXTERNAL_CATEGORY(xbt_parmap);
+
+namespace simgrid {
+namespace xbt {
+
+/** \addtogroup XBT_parmap
+ * \ingroup XBT_misc
+ * \brief Parallel map class
+ * \{
+ */
+template <typename T> class Parmap {
+public:
+ Parmap(unsigned num_workers, e_xbt_parmap_mode_t mode);
+ Parmap(const Parmap&) = delete;
+ ~Parmap();
+ void apply(void (*fun)(T), const std::vector<T>& data);
+ boost::optional<T> next();
+
+private:
+ enum Flag { PARMAP_WORK, PARMAP_DESTROY };
+
+ /**
+ * \brief Thread data transmission structure
+ */
+ class ThreadData {
+ public:
+ ThreadData(Parmap<T>& parmap, int id) : parmap(parmap), worker_id(id) {}
+ Parmap<T>& parmap;
+ int worker_id;
+ };
+
+ /**
+ * \brief Synchronization object (different specializations).
+ */
+ class Synchro {
+ public:
+ explicit Synchro(Parmap<T>& parmap) : parmap(parmap) {}
+ virtual ~Synchro() {}
+ /**
+ * \brief Wakes all workers and waits for them to finish the tasks.
+ *
+ * This function is called by the controller thread.
+ */
+ virtual void master_signal() = 0;
+ /**
+ * \brief Starts the parmap: waits for all workers to be ready and returns.
+ *
+ * This function is called by the controller thread.
+ */
+ virtual void master_wait() = 0;
+ /**
+ * \brief Ends the parmap: wakes the controller thread when all workers terminate.
+ *
+ * This function is called by all worker threads when they end (not including the controller).
+ */
+ virtual void worker_signal() = 0;
+ /**
+ * \brief Waits for some work to process.
+ *
+ * This function is called by each worker thread (not including the controller) when it has no more work to do.
+ *
+ * \param round the expected round number
+ */
+ virtual void worker_wait(unsigned) = 0;
+
+ protected:
+ Parmap<T>& parmap;
+ };
+
+ class PosixSynchro : public Synchro {
+ public:
+ explicit PosixSynchro(Parmap<T>& parmap);
+ ~PosixSynchro();
+ void master_signal();
+ void master_wait();
+ void worker_signal();
+ void worker_wait(unsigned round);
+
+ private:
+ xbt_os_cond_t ready_cond;
+ xbt_os_mutex_t ready_mutex;
+ xbt_os_cond_t done_cond;
+ xbt_os_mutex_t done_mutex;
+ };
+
+#if HAVE_FUTEX_H
+ class FutexSynchro : public Synchro {
+ public:
+ explicit FutexSynchro(Parmap<T>& parmap) : Synchro(parmap) {}
+ void master_signal();
+ void master_wait();
+ void worker_signal();
+ void worker_wait(unsigned);
+
+ private:
+ static void futex_wait(unsigned* uaddr, unsigned val);
+ static void futex_wake(unsigned* uaddr, unsigned val);
+ };
+#endif
+
+ class BusyWaitSynchro : public Synchro {
+ public:
+ explicit BusyWaitSynchro(Parmap<T>& parmap) : Synchro(parmap) {}
+ void master_signal();
+ void master_wait();
+ void worker_signal();
+ void worker_wait(unsigned);
+ };
+
+ static void* worker_main(void* arg);
+ Synchro* new_synchro(e_xbt_parmap_mode_t mode);
+ void work();
+
+ Flag status; /**< is the parmap active or being destroyed? */
+ unsigned work_round; /**< index of the current round */
+ xbt_os_thread_t* workers; /**< worker thread handlers */
+ unsigned num_workers; /**< total number of worker threads including the controller */
+ Synchro* synchro; /**< synchronization object */
+
+ unsigned thread_counter = 0; /**< number of workers that have done the work */
+ void (*fun)(const T) = nullptr; /**< function to run in parallel on each element of data */
+ const std::vector<T>* data = nullptr; /**< parameters to pass to fun in parallel */
+ std::atomic<unsigned> index; /**< index of the next element of data to pick */
+};
+
+/**
+ * \brief Creates a parallel map object
+ * \param num_workers number of worker threads to create
+ * \param mode how to synchronize the worker threads
+ */
+template <typename T> Parmap<T>::Parmap(unsigned num_workers, e_xbt_parmap_mode_t mode)
+{
+ XBT_CDEBUG(xbt_parmap, "Create new parmap (%u workers)", num_workers);
+
+ /* Initialize the thread pool data structure */
+ this->status = PARMAP_WORK;
+ this->work_round = 0;
+ this->workers = new xbt_os_thread_t[num_workers];
+ this->num_workers = num_workers;
+ this->synchro = new_synchro(mode);
+
+ /* Create the pool of worker threads */
+ this->workers[0] = nullptr;
+#if HAVE_PTHREAD_SETAFFINITY
+ int core_bind = 0;
+#endif
+ for (unsigned i = 1; i < num_workers; i++) {
+ ThreadData* data = new ThreadData(*this, i);
+ this->workers[i] = xbt_os_thread_create(nullptr, worker_main, data, nullptr);
+#if HAVE_PTHREAD_SETAFFINITY
+ xbt_os_thread_bind(this->workers[i], core_bind);
+ if (core_bind != xbt_os_get_numcores() - 1)
+ core_bind++;
+ else
+ core_bind = 0;
+#endif
+ }
+}
+
+/**
+ * \brief Destroys a parmap
+ */
+template <typename T> Parmap<T>::~Parmap()
+{
+ status = PARMAP_DESTROY;
+ synchro->master_signal();
+
+ for (unsigned i = 1; i < num_workers; i++)
+ xbt_os_thread_join(workers[i], nullptr);
+
+ delete[] workers;
+ delete synchro;
+}
+
+/**
+ * \brief Applies a list of tasks in parallel.
+ * \param fun the function to call in parallel
+ * \param data each element of this vector will be passed as an argument to fun
+ */
+template <typename T> void Parmap<T>::apply(void (*fun)(T), const std::vector<T>& data)
+{
+ /* Assign resources to worker threads (we are maestro here)*/
+ this->fun = fun;
+ this->data = &data;
+ this->index = 0;
+ this->synchro->master_signal(); // maestro runs futex_wait to wake all the minions (the working threads)
+ this->work(); // maestro works with its minions
+ this->synchro->master_wait(); // When there is no more work to do, then maestro waits for the last minion to stop
+ XBT_CDEBUG(xbt_parmap, "Job done"); // ... and proceeds
+}
+
+/**
+ * \brief Returns a next task to process.
+ *
+ * Worker threads call this function to get more work.
+ *
+ * \return the next task to process, or throws a std::out_of_range exception if there is no more work
+ */
+template <typename T> boost::optional<T> Parmap<T>::next()
+{
+ unsigned index = this->index++;
+ if (index < this->data->size())
+ return (*this->data)[index];
+ else
+ return boost::none;
+}
+
+/**
+ * \brief Main work loop: applies fun to elements in turn.
+ */
+template <typename T> void Parmap<T>::work()
+{
+ unsigned index = this->index++;
+ unsigned length = this->data->size();
+ while (index < length) {
+ this->fun((*this->data)[index]);
+ index = this->index++;
+ }
+}
+
+/**
+ * Get a synchronization object for given mode.
+ * \param mode the synchronization mode
+ */
+template <typename T> typename Parmap<T>::Synchro* Parmap<T>::new_synchro(e_xbt_parmap_mode_t mode)
+{
+ if (mode == XBT_PARMAP_DEFAULT) {
+#if HAVE_FUTEX_H
+ mode = XBT_PARMAP_FUTEX;
+#else
+ mode = XBT_PARMAP_POSIX;
+#endif
+ }
+ Synchro* res;
+ switch (mode) {
+ case XBT_PARMAP_POSIX:
+ res = new PosixSynchro(*this);
+ break;
+ case XBT_PARMAP_FUTEX:
+#if HAVE_FUTEX_H
+ res = new FutexSynchro(*this);
+#else
+ xbt_die("Fute is not available on this OS.");
+#endif
+ break;
+ case XBT_PARMAP_BUSY_WAIT:
+ res = new BusyWaitSynchro(*this);
+ break;
+ default:
+ THROW_IMPOSSIBLE;
+ }
+ return res;
+}
+
+/**
+ * \brief Main function of a worker thread.
+ */
+template <typename T> void* Parmap<T>::worker_main(void* arg)
+{
+ ThreadData* data = static_cast<ThreadData*>(arg);
+ Parmap<T>& parmap = data->parmap;
+ unsigned round = 0;
+ smx_context_t context = SIMIX_context_new(std::function<void()>(), nullptr, nullptr);
+ SIMIX_context_set_current(context);
+
+ XBT_CDEBUG(xbt_parmap, "New worker thread created");
+
+ /* Worker's main loop */
+ while (1) {
+ round++;
+ parmap.synchro->worker_wait(round);
+ if (parmap.status == PARMAP_DESTROY)
+ break;
+
+ XBT_CDEBUG(xbt_parmap, "Worker %d got a job", data->worker_id);
+ parmap.work();
+ parmap.synchro->worker_signal();
+ XBT_CDEBUG(xbt_parmap, "Worker %d has finished", data->worker_id);
+ }
+ /* We are destroying the parmap */
+ delete context;
+ delete data;
+ return nullptr;
+}
+
+template <typename T> Parmap<T>::PosixSynchro::PosixSynchro(Parmap<T>& parmap) : Synchro(parmap)
+{
+ ready_cond = xbt_os_cond_init();
+ ready_mutex = xbt_os_mutex_init();
+ done_cond = xbt_os_cond_init();
+ done_mutex = xbt_os_mutex_init();
+}
+
+template <typename T> Parmap<T>::PosixSynchro::~PosixSynchro()
+{
+ xbt_os_cond_destroy(ready_cond);
+ xbt_os_mutex_destroy(ready_mutex);
+ xbt_os_cond_destroy(done_cond);
+ xbt_os_mutex_destroy(done_mutex);
+}
+
+template <typename T> void Parmap<T>::PosixSynchro::master_signal()
+{
+ xbt_os_mutex_acquire(ready_mutex);
+ this->parmap.thread_counter = 1;
+ this->parmap.work_round++;
+ /* wake all workers */
+ xbt_os_cond_broadcast(ready_cond);
+ xbt_os_mutex_release(ready_mutex);
+}
+
+template <typename T> void Parmap<T>::PosixSynchro::master_wait()
+{
+ xbt_os_mutex_acquire(done_mutex);
+ if (this->parmap.thread_counter < this->parmap.num_workers) {
+ /* wait for all workers to be ready */
+ xbt_os_cond_wait(done_cond, done_mutex);
+ }
+ xbt_os_mutex_release(done_mutex);
+}
+
+template <typename T> void Parmap<T>::PosixSynchro::worker_signal()
+{
+ xbt_os_mutex_acquire(done_mutex);
+ this->parmap.thread_counter++;
+ if (this->parmap.thread_counter == this->parmap.num_workers) {
+ /* all workers have finished, wake the controller */
+ xbt_os_cond_signal(done_cond);
+ }
+ xbt_os_mutex_release(done_mutex);
+}
+
+template <typename T> void Parmap<T>::PosixSynchro::worker_wait(unsigned round)
+{
+ xbt_os_mutex_acquire(ready_mutex);
+ /* wait for more work */
+ if (this->parmap.work_round != round) {
+ xbt_os_cond_wait(ready_cond, ready_mutex);
+ }
+ xbt_os_mutex_release(ready_mutex);
+}
+
+#if HAVE_FUTEX_H
+template <typename T> inline void Parmap<T>::FutexSynchro::futex_wait(unsigned* uaddr, unsigned val)
+{
+ XBT_CVERB(xbt_parmap, "Waiting on futex %p", uaddr);
+ syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, nullptr, nullptr, 0);
+}
+
+template <typename T> inline void Parmap<T>::FutexSynchro::futex_wake(unsigned* uaddr, unsigned val)
+{
+ XBT_CVERB(xbt_parmap, "Waking futex %p", uaddr);
+ syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, nullptr, nullptr, 0);
+}
+
+template <typename T> void Parmap<T>::FutexSynchro::master_signal()
+{
+ this->parmap.thread_counter = 1;
+ __sync_add_and_fetch(&this->parmap.work_round, 1);
+ /* wake all workers */
+ futex_wake(&this->parmap.work_round, std::numeric_limits<int>::max());
+}
+
+template <typename T> void Parmap<T>::FutexSynchro::master_wait()
+{
+ unsigned count = this->parmap.thread_counter;
+ while (count < this->parmap.num_workers) {
+ /* wait for all workers to be ready */
+ futex_wait(&this->parmap.thread_counter, count);
+ count = this->parmap.thread_counter;
+ }
+}
+
+template <typename T> void Parmap<T>::FutexSynchro::worker_signal()
+{
+ unsigned count = __sync_add_and_fetch(&this->parmap.thread_counter, 1);
+ if (count == this->parmap.num_workers) {
+ /* all workers have finished, wake the controller */
+ futex_wake(&this->parmap.thread_counter, std::numeric_limits<int>::max());
+ }
+}
+
+template <typename T> void Parmap<T>::FutexSynchro::worker_wait(unsigned round)
+{
+ unsigned work_round = this->parmap.work_round;
+ /* wait for more work */
+ while (work_round != round) {
+ futex_wait(&this->parmap.work_round, work_round);
+ work_round = this->parmap.work_round;
+ }
+}
+#endif
+
+template <typename T> void Parmap<T>::BusyWaitSynchro::master_signal()
+{
+ this->parmap.thread_counter = 1;
+ __sync_add_and_fetch(&this->parmap.work_round, 1);
+}
+
+template <typename T> void Parmap<T>::BusyWaitSynchro::master_wait()
+{
+ while (this->parmap.thread_counter < this->parmap.num_workers) {
+ xbt_os_thread_yield();
+ }
+}
+
+template <typename T> void Parmap<T>::BusyWaitSynchro::worker_signal()
+{
+ __sync_add_and_fetch(&this->parmap.thread_counter, 1);
+}
+
+template <typename T> void Parmap<T>::BusyWaitSynchro::worker_wait(unsigned round)
+{
+ /* wait for more work */
+ while (this->parmap.work_round != round) {
+ xbt_os_thread_yield();
+ }
+}
+
+/** \} */
+}
+}
+
+#endif
-/* Copyright (c) 2010-2015. The SimGrid Team.
- * All rights reserved. */
+/* Copyright (c) 2010-2017. The SimGrid Team. All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
#include "src/instr/instr_private.h"
#include "surf/surf.h"
+#include <string>
#include <vector>
XBT_LOG_NEW_CATEGORY(instr, "Logging the behavior of the tracing system (used for Visualization/Analysis of simulations)");
}
trace_active = 1;
XBT_DEBUG("Tracing is on");
-
- /* other trace initialization */
- created_categories = xbt_dict_new_homogeneous(xbt_free_f);
- declared_marks = xbt_dict_new_homogeneous(xbt_free_f);
- user_host_variables = xbt_dict_new_homogeneous(xbt_free_f);
- user_vm_variables = xbt_dict_new_homogeneous(xbt_free_f);
- user_link_variables = xbt_dict_new_homogeneous(xbt_free_f);
}
return 0;
}
PJ_container_release();
rootType = nullptr;
- xbt_dict_free(&user_link_variables);
- xbt_dict_free(&user_host_variables);
- xbt_dict_free(&user_vm_variables);
- xbt_dict_free(&declared_marks);
- xbt_dict_free(&created_categories);
-
/* close the trace files */
const char* format = xbt_cfg_get_string(OPT_TRACING_FORMAT);
XBT_DEBUG("Tracing format %s\n", format);
xbt_dynar_free (&types);
}
-static void output_categories (const char *name, xbt_dynar_t cats, FILE *file)
+static void output_categories(const char* name, FILE* file)
{
- unsigned int i;
+ unsigned int i = created_categories.size();
fprintf (file, " values = (");
- for (i = xbt_dynar_length(cats); i > 0; i--) {
- char *cat = *(static_cast<char**>(xbt_dynar_get_ptr(cats, i - 1)));
- fprintf (file, "\"%s%s\"", name, cat);
- if (i - 1 > 0){
+ for (auto cat : created_categories) {
+ --i;
+ fprintf(file, "\"%s%s\"", name, cat.c_str());
+ if (i > 0) {
fprintf (file, ",");
}else{
fprintf (file, ");\n");
}
}
- xbt_dynar_free (&cats);
}
static void uncat_configuration (FILE *file)
" host = {\n"
" type = \"square\";\n"
" size = \"power\";\n");
- output_categories ("p", TRACE_get_categories(), file);
+ output_categories("p", file);
fprintf (file,
" };\n"
" link = {\n"
" type = \"rhombus\";\n"
" size = \"bandwidth\";\n");
- output_categories ("b", TRACE_get_categories(), file);
+ output_categories("b", file);
fprintf (file, " };\n");
//close
}
{
if (output && strlen(output) > 0){
//check if we do have categories declared
- if (xbt_dict_is_empty(created_categories)){
+ if (created_categories.empty()) {
XBT_INFO("No categories declared, ignoring generation of %s graph configuration", name);
return;
}
#include "src/surf/network_interface.hpp"
#include "src/surf/surf_private.h"
#include "surf/surf.h"
+#include <algorithm>
typedef enum {
INSTR_US_DECLARE,
XBT_LOG_NEW_DEFAULT_SUBCATEGORY (instr_api, instr, "API");
-xbt_dict_t created_categories = nullptr;
-xbt_dict_t declared_marks = nullptr;
-xbt_dict_t user_host_variables = nullptr;
-xbt_dict_t user_vm_variables = nullptr;
-xbt_dict_t user_link_variables = nullptr;
-extern xbt_dict_t trivaNodeTypes;
-extern xbt_dict_t trivaEdgeTypes;
+std::set<std::string> created_categories;
+std::set<std::string> declared_marks;
+std::set<std::string> user_host_variables;
+std::set<std::string> user_vm_variables;
+std::set<std::string> user_link_variables;
+extern std::set<std::string> trivaNodeTypes;
+extern std::set<std::string> trivaEdgeTypes;
-static xbt_dynar_t instr_dict_to_dynar (xbt_dict_t filter)
+static xbt_dynar_t instr_set_to_dynar(std::set<std::string>* filter)
{
if (not TRACE_is_enabled() || not TRACE_needs_platform())
return nullptr;
xbt_dynar_t ret = xbt_dynar_new (sizeof(char*), &xbt_free_ref);
- xbt_dict_cursor_t cursor = nullptr;
- char *name;
- char *value;
- xbt_dict_foreach(filter, cursor, name, value) {
- xbt_dynar_push_as (ret, char*, xbt_strdup(name));
- }
+ for (auto name : *filter)
+ xbt_dynar_push_as(ret, char*, xbt_strdup(name.c_str()));
+
return ret;
}
return;
//check if category is already created
- if (xbt_dict_get_or_null(created_categories, category) != nullptr)
+ if (created_categories.find(category) != created_categories.end())
return;
-
- xbt_dict_set (created_categories, category, xbt_strdup("1"), nullptr);
+ else
+ created_categories.insert(category);
//define final_color
char final_color[INSTR_DEFAULT_STR_SIZE];
{
if (not TRACE_is_enabled() || not TRACE_categorized())
return nullptr;
-
- return instr_dict_to_dynar (created_categories);
+ return instr_set_to_dynar(&created_categories);
}
/** \ingroup TRACE_mark
THROWF (tracing_error, 1, "mark_type is nullptr");
//check if mark_type is already declared
- if (xbt_dict_get_or_null(declared_marks, mark_type) != nullptr) {
+ if (declared_marks.find(mark_type) != declared_marks.end()) {
THROWF (tracing_error, 1, "mark_type with name (%s) is already declared", mark_type);
}
XBT_DEBUG("MARK,declare %s", mark_type);
PJ_type_event_new(mark_type, PJ_type_get_root());
- xbt_dict_set (declared_marks, mark_type, xbt_strdup("1"), nullptr);
+ declared_marks.insert(mark_type);
}
/** \ingroup TRACE_mark
if (not TRACE_is_enabled())
return nullptr;
- return instr_dict_to_dynar (declared_marks);
+ return instr_set_to_dynar(&declared_marks);
}
-static void instr_user_variable(double time, const char *resource, const char *variable, const char *father_type,
- double value, InstrUserVariable what, const char *color, xbt_dict_t filter)
+static void instr_user_variable(double time, const char* resource, const char* variable, const char* father_type,
+ double value, InstrUserVariable what, const char* color, std::set<std::string>* filter)
{
/* safe switches. tracing has to be activated and if platform is not traced, we don't allow user variables */
if (not TRACE_is_enabled() || not TRACE_needs_platform())
return;
//check if variable is already declared
- char *created = (char*)xbt_dict_get_or_null(filter, variable);
+ auto created = filter->find(variable);
if (what == INSTR_US_DECLARE){
- if (not created) { // not declared yet
- xbt_dict_set (filter, variable, xbt_strdup("1"), nullptr);
+ if (created == filter->end()) { // not declared yet
+ filter->insert(variable);
instr_new_user_variable_type (father_type, variable, color);
}
}else{
- if (created) { // declared, let's work
+ if (created != filter->end()) { // declared, let's work
char valuestr[100];
snprintf(valuestr, 100, "%g", value);
container_t container = PJ_container_get(resource);
std::vector<simgrid::surf::LinkImpl*> route;
simgrid::kernel::routing::NetZoneImpl::getGlobalRoute(src_elm, dst_elm, &route, nullptr);
for (auto link : route)
- instr_user_variable(time, link->cname(), variable, father_type, value, what, nullptr, user_link_variables);
+ instr_user_variable(time, link->cname(), variable, father_type, value, what, nullptr, &user_link_variables);
}
/** \ingroup TRACE_API
*/
void TRACE_vm_variable_declare (const char *variable)
{
- instr_user_variable(0, nullptr, variable, "MSG_VM", 0, INSTR_US_DECLARE, nullptr, user_vm_variables);
+ instr_user_variable(0, nullptr, variable, "MSG_VM", 0, INSTR_US_DECLARE, nullptr, &user_vm_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_vm_variable_declare_with_color (const char *variable, const char *color)
{
- instr_user_variable(0, nullptr, variable, "MSG_VM", 0, INSTR_US_DECLARE, color, user_vm_variables);
+ instr_user_variable(0, nullptr, variable, "MSG_VM", 0, INSTR_US_DECLARE, color, &user_vm_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_vm_variable_set_with_time (double time, const char *vm, const char *variable, double value)
{
- instr_user_variable(time, vm, variable, "MSG_VM", value, INSTR_US_SET, nullptr, user_vm_variables);
+ instr_user_variable(time, vm, variable, "MSG_VM", value, INSTR_US_SET, nullptr, &user_vm_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_vm_variable_add_with_time (double time, const char *vm, const char *variable, double value)
{
- instr_user_variable(time, vm, variable, "MSG_VM", value, INSTR_US_ADD, nullptr, user_vm_variables);
+ instr_user_variable(time, vm, variable, "MSG_VM", value, INSTR_US_ADD, nullptr, &user_vm_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_vm_variable_sub_with_time (double time, const char *vm, const char *variable, double value)
{
- instr_user_variable(time, vm, variable, "MSG_VM", value, INSTR_US_SUB, nullptr, user_vm_variables);
+ instr_user_variable(time, vm, variable, "MSG_VM", value, INSTR_US_SUB, nullptr, &user_vm_variables);
}
/* for host variables */
*/
void TRACE_host_variable_declare (const char *variable)
{
- instr_user_variable(0, nullptr, variable, "HOST", 0, INSTR_US_DECLARE, nullptr, user_host_variables);
+ instr_user_variable(0, nullptr, variable, "HOST", 0, INSTR_US_DECLARE, nullptr, &user_host_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_host_variable_declare_with_color (const char *variable, const char *color)
{
- instr_user_variable(0, nullptr, variable, "HOST", 0, INSTR_US_DECLARE, color, user_host_variables);
+ instr_user_variable(0, nullptr, variable, "HOST", 0, INSTR_US_DECLARE, color, &user_host_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_host_variable_set_with_time (double time, const char *host, const char *variable, double value)
{
- instr_user_variable(time, host, variable, "HOST", value, INSTR_US_SET, nullptr, user_host_variables);
+ instr_user_variable(time, host, variable, "HOST", value, INSTR_US_SET, nullptr, &user_host_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_host_variable_add_with_time (double time, const char *host, const char *variable, double value)
{
- instr_user_variable(time, host, variable, "HOST", value, INSTR_US_ADD, nullptr, user_host_variables);
+ instr_user_variable(time, host, variable, "HOST", value, INSTR_US_ADD, nullptr, &user_host_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_host_variable_sub_with_time (double time, const char *host, const char *variable, double value)
{
- instr_user_variable(time, host, variable, "HOST", value, INSTR_US_SUB, nullptr, user_host_variables);
+ instr_user_variable(time, host, variable, "HOST", value, INSTR_US_SUB, nullptr, &user_host_variables);
}
/** \ingroup TRACE_user_variables
*/
xbt_dynar_t TRACE_get_host_variables ()
{
- return instr_dict_to_dynar (user_host_variables);
+ return instr_set_to_dynar(&user_host_variables);
}
/* for link variables */
*/
void TRACE_link_variable_declare (const char *variable)
{
- instr_user_variable (0, nullptr, variable, "LINK", 0, INSTR_US_DECLARE, nullptr, user_link_variables);
+ instr_user_variable(0, nullptr, variable, "LINK", 0, INSTR_US_DECLARE, nullptr, &user_link_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_link_variable_declare_with_color (const char *variable, const char *color)
{
- instr_user_variable (0, nullptr, variable, "LINK", 0, INSTR_US_DECLARE, color, user_link_variables);
+ instr_user_variable(0, nullptr, variable, "LINK", 0, INSTR_US_DECLARE, color, &user_link_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_link_variable_set_with_time (double time, const char *link, const char *variable, double value)
{
- instr_user_variable (time, link, variable, "LINK", value, INSTR_US_SET, nullptr, user_link_variables);
+ instr_user_variable(time, link, variable, "LINK", value, INSTR_US_SET, nullptr, &user_link_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_link_variable_add_with_time (double time, const char *link, const char *variable, double value)
{
- instr_user_variable (time, link, variable, "LINK", value, INSTR_US_ADD, nullptr, user_link_variables);
+ instr_user_variable(time, link, variable, "LINK", value, INSTR_US_ADD, nullptr, &user_link_variables);
}
/** \ingroup TRACE_user_variables
*/
void TRACE_link_variable_sub_with_time (double time, const char *link, const char *variable, double value)
{
- instr_user_variable (time, link, variable, "LINK", value, INSTR_US_SUB, nullptr, user_link_variables);
+ instr_user_variable(time, link, variable, "LINK", value, INSTR_US_SUB, nullptr, &user_link_variables);
}
/* for link variables, but with src and dst used for get_route */
*/
xbt_dynar_t TRACE_get_link_variables ()
{
- return instr_dict_to_dynar (user_link_variables);
+ return instr_set_to_dynar(&user_link_variables);
}
/** \ingroup TRACE_user_variables
*/
xbt_dynar_t TRACE_get_node_types ()
{
- return instr_dict_to_dynar (trivaNodeTypes);
+ return instr_set_to_dynar(&trivaNodeTypes);
}
/** \ingroup TRACE_API
*/
xbt_dynar_t TRACE_get_edge_types ()
{
- return instr_dict_to_dynar (trivaEdgeTypes);
+ return instr_set_to_dynar(&trivaEdgeTypes);
}
static container_t rootContainer = nullptr; /* the root container */
static xbt_dict_t allContainers = nullptr; /* all created containers indexed by name */
-xbt_dict_t trivaNodeTypes = nullptr; /* all host types defined */
-xbt_dict_t trivaEdgeTypes = nullptr; /* all link types defined */
+std::set<std::string> trivaNodeTypes; /* all host types defined */
+std::set<std::string> trivaEdgeTypes; /* all link types defined */
long long int instr_new_paje_id ()
{
void PJ_container_alloc ()
{
allContainers = xbt_dict_new_homogeneous(nullptr);
- trivaNodeTypes = xbt_dict_new_homogeneous(xbt_free_f);
- trivaEdgeTypes = xbt_dict_new_homogeneous(xbt_free_f);
}
void PJ_container_release ()
{
xbt_dict_free (&allContainers);
- xbt_dict_free (&trivaNodeTypes);
- xbt_dict_free (&trivaEdgeTypes);
}
void PJ_container_set_root (container_t root)
//register NODE types for triva configuration
if (newContainer->kind == INSTR_HOST || newContainer->kind == INSTR_LINK || newContainer->kind == INSTR_ROUTER) {
- xbt_dict_set (trivaNodeTypes, newContainer->type->name, xbt_strdup("1"), nullptr);
+ trivaNodeTypes.insert(newContainer->type->name);
}
return newContainer;
}
#include <xbt/base.h>
-#include "simgrid/instr.h"
#include "instr/instr_interface.h"
-#include "src/internal_config.h"
+#include "simgrid/instr.h"
#include "simgrid_config.h"
+#include "src/internal_config.h"
+#include <set>
SG_BEGIN_DECL()
};
-
-extern XBT_PRIVATE xbt_dict_t created_categories;
-extern XBT_PRIVATE xbt_dict_t declared_marks;
-extern XBT_PRIVATE xbt_dict_t user_host_variables;
-extern XBT_PRIVATE xbt_dict_t user_vm_variables;
-extern XBT_PRIVATE xbt_dict_t user_link_variables;
+extern XBT_PRIVATE std::set<std::string> created_categories;
+extern XBT_PRIVATE std::set<std::string> declared_marks;
+extern XBT_PRIVATE std::set<std::string> user_host_variables;
+extern XBT_PRIVATE std::set<std::string> user_vm_variables;
+extern XBT_PRIVATE std::set<std::string> user_link_variables;
extern XBT_PRIVATE double TRACE_last_timestamp_to_dump;
/* instr_paje_header.c */
XBT_PUBLIC(void) TRACE_surf_resource_utilization_alloc();
/* instr_paje.c */
-extern XBT_PRIVATE xbt_dict_t trivaNodeTypes;
-extern XBT_PRIVATE xbt_dict_t trivaEdgeTypes;
+extern XBT_PRIVATE std::set<std::string> trivaNodeTypes;
+extern XBT_PRIVATE std::set<std::string> trivaEdgeTypes;
XBT_PRIVATE long long int instr_new_paje_id ();
XBT_PRIVATE void PJ_container_alloc ();
XBT_PRIVATE void PJ_container_release ();
-/* Copyright (c) 2010-2015. The SimGrid Team.
+/* Copyright (c) 2010-2017. The SimGrid Team.
* All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
#include "src/instr/instr_private.h"
+#include <string>
+#include <unordered_map>
XBT_LOG_NEW_DEFAULT_SUBCATEGORY (instr_resource, instr, "tracing (un)-categorized resource utilization");
//to check if variables were previously set to 0, otherwise paje won't simulate them
-static xbt_dict_t platform_variables;
+static std::unordered_map<std::string, std::string> platform_variables;
//used by all methods
static void __TRACE_surf_check_variable_set_to_zero(double now, const char *variable, const char *resource)
*/
// create a key considering the resource and variable
- int n = strlen(variable)+strlen(resource)+1;
- char *key = (char*)xbt_malloc(n*sizeof(char));
- snprintf (key, n, "%s%s", resource, variable);
+ std::string key = std::string(resource) + variable;
// check if key exists: if it doesn't, set the variable to zero and mark this in the dict
- if (not xbt_dict_get_or_null(platform_variables, key)) {
+ if (platform_variables.find(key) == platform_variables.end()) {
container_t container = PJ_container_get (resource);
type_t type = PJ_type_get (variable, container->type);
new SetVariableEvent (now, container, type, 0);
- xbt_dict_set(platform_variables, key, (char*)"", nullptr);
+ platform_variables[key] = std::string("");
}
- xbt_free(key);
}
static void instr_event (double now, double delta, type_t variable, container_t resource, double value)
instr_event (now, delta, type, container, value);
}
}
-
-void TRACE_surf_resource_utilization_alloc()
-{
- platform_variables = xbt_dict_new_homogeneous(nullptr);
-}
-
-void TRACE_surf_resource_utilization_release()
-{
- xbt_dict_free(&platform_variables);
-}
XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_mailbox, simix, "Mailbox implementation");
-static xbt_dict_t mailboxes = xbt_dict_new_homogeneous([](void* data) {
- delete static_cast<smx_mailbox_t>(data);
-});
+static std::map<std::string, smx_mailbox_t>* mailboxes = new std::map<std::string, smx_mailbox_t>;
void SIMIX_mailbox_exit()
{
- xbt_dict_free(&mailboxes);
+ for (auto elm : *mailboxes)
+ delete elm.second;
+ delete mailboxes;
}
/******************************************************************************/
/** @brief Returns the mailbox of that name, or nullptr */
MailboxImpl* MailboxImpl::byNameOrNull(const char* name)
{
- return static_cast<smx_mailbox_t>(xbt_dict_get_or_null(mailboxes, name));
+ auto mbox = mailboxes->find(name);
+ if (mbox != mailboxes->end())
+ return mbox->second;
+ else
+ return nullptr;
}
/** @brief Returns the mailbox of that name, newly created on need */
MailboxImpl* MailboxImpl::byNameOrCreate(const char* name)
{
xbt_assert(name, "Mailboxes must have a name");
/* two processes may have pushed the same mbox_create simcall at the same time */
- smx_mailbox_t mbox = static_cast<smx_mailbox_t>(xbt_dict_get_or_null(mailboxes, name));
- if (not mbox) {
- mbox = new MailboxImpl(name);
+ auto m = mailboxes->find(name);
+ if (m == mailboxes->end()) {
+ smx_mailbox_t mbox = new MailboxImpl(name);
XBT_DEBUG("Creating a mailbox at %p with name %s", mbox, name);
- xbt_dict_set(mailboxes, mbox->name_, mbox, nullptr);
- }
- return mbox;
+ (*mailboxes)[mbox->name_] = mbox;
+ return mbox;
+ } else
+ return m->second;
}
/** @brief set the receiver of the mailbox to allow eager sends
* \param actor The receiving dude
XBT_PRIVATE void SIMIX_context_mod_init();
XBT_PRIVATE void SIMIX_context_mod_exit();
-XBT_PRIVATE smx_context_t SIMIX_context_new(
- std::function<void()> code,
- void_pfn_smxprocess_t cleanup_func,
- smx_actor_t simix_process);
+XBT_PUBLIC(smx_context_t)
+SIMIX_context_new(std::function<void()> code, void_pfn_smxprocess_t cleanup_func, smx_actor_t simix_process);
#ifndef WIN32
XBT_PUBLIC_DATA(char sigsegv_stack[SIGSTKSZ]);
XBT_PRIVATE void *SIMIX_context_stack_new();
XBT_PRIVATE void SIMIX_context_stack_delete(void *stack);
-XBT_PRIVATE void SIMIX_context_set_current(smx_context_t context);
+XBT_PUBLIC(void) SIMIX_context_set_current(smx_context_t context);
XBT_PRIVATE smx_context_t SIMIX_context_get_current();
XBT_PUBLIC(int) SIMIX_process_get_maxpid();
-/* Copyright (c) 2015. The SimGrid Team. All rights reserved. */
+/* Copyright (c) 2015-2017. The SimGrid Team. All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
// BoostContextFactory
-bool BoostContext::parallel_ = false;
-xbt_parmap_t BoostContext::parmap_ = nullptr;
-uintptr_t BoostContext::threads_working_ = 0;
+bool BoostContext::parallel_ = false;
+simgrid::xbt::Parmap<smx_actor_t>* BoostContext::parmap_ = nullptr;
+uintptr_t BoostContext::threads_working_ = 0;
xbt_os_thread_key_t BoostContext::worker_id_key_;
-unsigned long BoostContext::process_index_ = 0;
-BoostContext* BoostContext::maestro_context_ = nullptr;
+unsigned long BoostContext::process_index_ = 0;
+BoostContext* BoostContext::maestro_context_ = nullptr;
std::vector<BoostContext*> BoostContext::workers_context_;
BoostContextFactory::BoostContextFactory()
if (BoostContext::parallel_) {
#if HAVE_THREAD_CONTEXTS
int nthreads = SIMIX_context_get_nthreads();
- BoostContext::parmap_ = xbt_parmap_new(nthreads, SIMIX_context_get_parallel_mode());
+ BoostContext::parmap_ = new simgrid::xbt::Parmap<smx_actor_t>(nthreads, SIMIX_context_get_parallel_mode());
BoostContext::workers_context_.clear();
BoostContext::workers_context_.resize(nthreads, nullptr);
BoostContext::maestro_context_ = nullptr;
{
#if HAVE_THREAD_CONTEXTS
if (BoostContext::parmap_) {
- xbt_parmap_destroy(BoostContext::parmap_);
+ delete BoostContext::parmap_;
BoostContext::parmap_ = nullptr;
}
BoostContext::workers_context_.clear();
#if HAVE_THREAD_CONTEXTS
if (BoostContext::parallel_) {
BoostContext::threads_working_ = 0;
- xbt_parmap_apply(BoostContext::parmap_,
- [](void* arg) {
- smx_actor_t process = static_cast<smx_actor_t>(arg);
- BoostContext* context = static_cast<BoostContext*>(process->context);
- return context->resume();
- },
- simix_global->process_to_run);
+ BoostContext::parmap_->apply(
+ [](smx_actor_t process) {
+ BoostContext* context = static_cast<BoostContext*>(process->context);
+ return context->resume();
+ },
+ simix_global->process_to_run);
} else
#endif
{
- if (xbt_dynar_is_empty(simix_global->process_to_run))
+ if (simix_global->process_to_run.empty())
return;
- smx_actor_t first_process = xbt_dynar_get_as(simix_global->process_to_run, 0, smx_actor_t);
+ smx_actor_t first_process = simix_global->process_to_run.front();
BoostContext::process_index_ = 1;
/* execute the first process */
static_cast<BoostContext*>(first_process->context)->resume();
unsigned long int i = process_index_;
process_index_++;
- if (i < xbt_dynar_length(simix_global->process_to_run)) {
+ if (i < simix_global->process_to_run.size()) {
/* execute the next process */
XBT_DEBUG("Run next process");
- next_context =
- static_cast<BoostSerialContext*>(xbt_dynar_get_as(simix_global->process_to_run, i, smx_actor_t)->context);
+ next_context = static_cast<BoostSerialContext*>(simix_global->process_to_run[i]->context);
} else {
/* all processes were run, return to maestro */
XBT_DEBUG("No more process to run");
void BoostParallelContext::suspend()
{
- smx_actor_t next_work = static_cast<smx_actor_t>(xbt_parmap_next(parmap_));
- BoostParallelContext* next_context = nullptr;
-
- if (next_work != nullptr) {
+ boost::optional<smx_actor_t> next_work = parmap_->next();
+ BoostParallelContext* next_context;
+ if (next_work) {
XBT_DEBUG("Run next process");
- next_context = static_cast<BoostParallelContext*>(next_work->context);
+ next_context = static_cast<BoostParallelContext*>(next_work.get()->context);
} else {
XBT_DEBUG("No more processes to run");
uintptr_t worker_id = (uintptr_t)xbt_os_thread_get_specific(worker_id_key_);
-/* Copyright (c) 2015. The SimGrid Team. All rights reserved. */
+/* Copyright (c) 2015-2017. The SimGrid Team. All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
#include <functional>
#include <vector>
-#include <xbt/parmap.h>
+#include <xbt/parmap.hpp>
#include <simgrid/simix.hpp>
class BoostContext : public Context {
protected: // static
static bool parallel_;
- static xbt_parmap_t parmap_;
+ static simgrid::xbt::Parmap<smx_actor_t>* parmap_;
static std::vector<BoostContext*> workers_context_;
static uintptr_t threads_working_;
static xbt_os_thread_key_t worker_id_key_;
#include "src/internal_config.h"
-#include "xbt/parmap.h"
+#include "xbt/parmap.hpp"
#include "src/simix/smx_private.h"
#include "mc/mc.h"
// ***** Loads of static stuff
#if HAVE_THREAD_CONTEXTS
-static xbt_parmap_t raw_parmap;
+static simgrid::xbt::Parmap<smx_actor_t>* raw_parmap;
static simgrid::kernel::context::RawContext** raw_workers_context; /* space to save the worker context in each thread */
static uintptr_t raw_threads_working; /* number of threads that have started their work */
static xbt_os_thread_key_t raw_worker_id_key; /* thread-specific storage for the thread id */
RawContextFactory::~RawContextFactory()
{
#if HAVE_THREAD_CONTEXTS
- if (raw_parmap)
- xbt_parmap_destroy(raw_parmap);
+ delete raw_parmap;
xbt_free(raw_workers_context);
#endif
}
void RawContextFactory::run_all_serial()
{
- if (xbt_dynar_is_empty(simix_global->process_to_run))
+ if (simix_global->process_to_run.empty())
return;
- smx_actor_t first_process =
- xbt_dynar_get_as(simix_global->process_to_run, 0, smx_actor_t);
+ smx_actor_t first_process = simix_global->process_to_run.front();
raw_process_index = 1;
static_cast<RawContext*>(first_process->context)->resume_serial();
}
#if HAVE_THREAD_CONTEXTS
raw_threads_working = 0;
if (raw_parmap == nullptr)
- raw_parmap = xbt_parmap_new(
- SIMIX_context_get_nthreads(), SIMIX_context_get_parallel_mode());
- xbt_parmap_apply(raw_parmap,
- [](void* arg) {
- smx_actor_t process = static_cast<smx_actor_t>(arg);
+ raw_parmap = new simgrid::xbt::Parmap<smx_actor_t>(SIMIX_context_get_nthreads(), SIMIX_context_get_parallel_mode());
+ raw_parmap->apply(
+ [](smx_actor_t process) {
RawContext* context = static_cast<RawContext*>(process->context);
context->resume_parallel();
},
RawContext* next_context = nullptr;
unsigned long int i = raw_process_index;
raw_process_index++;
- if (i < xbt_dynar_length(simix_global->process_to_run)) {
+ if (i < simix_global->process_to_run.size()) {
/* execute the next process */
XBT_DEBUG("Run next process");
- next_context = static_cast<RawContext*>(xbt_dynar_get_as(simix_global->process_to_run, i, smx_actor_t)->context);
+ next_context = static_cast<RawContext*>(simix_global->process_to_run[i]->context);
} else {
/* all processes were run, return to maestro */
XBT_DEBUG("No more process to run");
{
#if HAVE_THREAD_CONTEXTS
/* determine the next context */
- smx_actor_t next_work = static_cast<smx_actor_t>(xbt_parmap_next(raw_parmap));
- RawContext* next_context = nullptr;
-
- if (next_work != nullptr) {
+ boost::optional<smx_actor_t> next_work = raw_parmap->next();
+ RawContext* next_context;
+ if (next_work) {
/* there is a next process to resume */
XBT_DEBUG("Run next process");
- next_context = static_cast<RawContext*>(next_work->context);
+ next_context = static_cast<RawContext*>(next_work.get()->context);
} else {
/* all processes were run, go to the barrier */
XBT_DEBUG("No more processes to run");
/** @brief Resumes all processes ready to run. */
void RawContextFactory::run_all_adaptative()
{
- unsigned long nb_processes = xbt_dynar_length(simix_global->process_to_run);
+ unsigned long nb_processes = simix_global->process_to_run.size();
if (SIMIX_context_is_parallel() &&
static_cast<unsigned long>(SIMIX_context_get_parallel_threshold()) < nb_processes) {
raw_context_parallel = true;
{
if (smx_ctx_thread_sem == nullptr) {
// Serial execution
- smx_actor_t process;
- unsigned int cursor;
- xbt_dynar_foreach(simix_global->process_to_run, cursor, process) {
+ for (smx_actor_t process : simix_global->process_to_run) {
XBT_DEBUG("Handling %p",process);
ThreadContext* context = static_cast<ThreadContext*>(process->context);
xbt_os_sem_release(context->begin_);
}
} else {
// Parallel execution
- unsigned int index;
- smx_actor_t process;
- xbt_dynar_foreach(simix_global->process_to_run, index, process)
+ for (smx_actor_t process : simix_global->process_to_run)
xbt_os_sem_release(static_cast<ThreadContext*>(process->context)->begin_);
- xbt_dynar_foreach(simix_global->process_to_run, index, process)
- xbt_os_sem_acquire(static_cast<ThreadContext*>(process->context)->end_);
+ for (smx_actor_t process : simix_global->process_to_run)
+ xbt_os_sem_acquire(static_cast<ThreadContext*>(process->context)->end_);
}
}
#include <ucontext.h> /* context relative declarations */
-#include "src/simix/ActorImpl.hpp"
-#include "src/simix/smx_private.h"
-#include "xbt/parmap.h"
#include "mc/mc.h"
#include "src/mc/mc_ignore.h"
-
+#include "src/simix/ActorImpl.hpp"
+#include "src/simix/smx_private.h"
+#include "xbt/parmap.hpp"
/** Many integers are needed to store a pointer
*
}}}
#if HAVE_THREAD_CONTEXTS
-static xbt_parmap_t sysv_parmap;
+static simgrid::xbt::Parmap<smx_actor_t>* sysv_parmap;
static simgrid::kernel::context::ParallelUContext** sysv_workers_context; /* space to save the worker's context in each thread */
static uintptr_t sysv_threads_working; /* number of threads that have started their work */
static xbt_os_thread_key_t sysv_worker_id_key; /* thread-specific storage for the thread id */
UContextFactory::~UContextFactory()
{
#if HAVE_THREAD_CONTEXTS
- if (sysv_parmap)
- xbt_parmap_destroy(sysv_parmap);
+ delete sysv_parmap;
xbt_free(sysv_workers_context);
#endif
}
// with simix_global->context_factory (which might not be initialized
// when bootstrapping):
if (sysv_parmap == nullptr)
- sysv_parmap = xbt_parmap_new(
- SIMIX_context_get_nthreads(), SIMIX_context_get_parallel_mode());
-
- xbt_parmap_apply(sysv_parmap,
- [](void* arg) {
- smx_actor_t process = (smx_actor_t) arg;
- ParallelUContext* context = static_cast<ParallelUContext*>(process->context);
- context->resume();
- },
- simix_global->process_to_run);
+ sysv_parmap =
+ new simgrid::xbt::Parmap<smx_actor_t>(SIMIX_context_get_nthreads(), SIMIX_context_get_parallel_mode());
+
+ sysv_parmap->apply(
+ [](smx_actor_t process) {
+ ParallelUContext* context = static_cast<ParallelUContext*>(process->context);
+ context->resume();
+ },
+ simix_global->process_to_run);
#else
xbt_die("You asked for a parallel execution, but you don't have any threads.");
#endif
} else {
// Serial:
- if (xbt_dynar_is_empty(simix_global->process_to_run))
+ if (simix_global->process_to_run.empty())
return;
- smx_actor_t first_process = xbt_dynar_get_as(simix_global->process_to_run, 0, smx_actor_t);
+ smx_actor_t first_process = simix_global->process_to_run.front();
sysv_process_index = 1;
SerialUContext* context = static_cast<SerialUContext*>(first_process->context);
context->resume();
SerialUContext* next_context = nullptr;
unsigned long int i = sysv_process_index++;
- if (i < xbt_dynar_length(simix_global->process_to_run)) {
+ if (i < simix_global->process_to_run.size()) {
/* execute the next process */
XBT_DEBUG("Run next process");
- next_context = (SerialUContext*) xbt_dynar_get_as(
- simix_global->process_to_run,i, smx_actor_t)->context;
+ next_context = static_cast<SerialUContext*>(simix_global->process_to_run[i]->context);
} else {
/* all processes were run, return to maestro */
XBT_DEBUG("No more process to run");
- next_context = (SerialUContext*) sysv_maestro_context;
+ next_context = static_cast<SerialUContext*>(sysv_maestro_context);
}
SIMIX_context_set_current(next_context);
swapcontext(&this->uc_, &next_context->uc_);
void SerialUContext::resume()
{
SIMIX_context_set_current(this);
- swapcontext(&((SerialUContext*)sysv_maestro_context)->uc_, &this->uc_);
+ swapcontext(&static_cast<SerialUContext*>(sysv_maestro_context)->uc_, &this->uc_);
}
void ParallelUContext::stop()
// Store the number of my containing body in os-thread-specific area :
xbt_os_thread_set_specific(sysv_worker_id_key, (void*) worker_id);
// Get my current soul:
- ParallelUContext* worker_context = (ParallelUContext*) SIMIX_context_self();
+ ParallelUContext* worker_context = static_cast<ParallelUContext*>(SIMIX_context_self());
// Write down that this soul is hosted in that body (for now)
sysv_workers_context[worker_id] = worker_context;
// Retrieve the system-level info that fuels this soul:
- ucontext_t* worker_stack = &((ParallelUContext*) worker_context)->uc_;
+ ucontext_t* worker_stack = &worker_context->uc_;
// Write in simix that I switched my soul
SIMIX_context_set_current(this);
// Actually do that using the relevant library call:
#if HAVE_THREAD_CONTEXTS
/* determine the next context */
// Get the next soul to embody now:
- smx_actor_t next_work = (smx_actor_t) xbt_parmap_next(sysv_parmap);
- ParallelUContext* next_context = nullptr;
- // Will contain the next soul to run, either simulated or initial minion's one
- ucontext_t* next_stack;
-
- if (next_work != nullptr) {
+ boost::optional<smx_actor_t> next_work = sysv_parmap->next();
+ ParallelUContext* next_context;
+ if (next_work) {
// There is a next soul to embody (ie, a next process to resume)
XBT_DEBUG("Run next process");
- next_context = (ParallelUContext*) next_work->context;
+ next_context = static_cast<ParallelUContext*>(next_work.get()->context);
} else {
// All processes were run, go to the barrier
XBT_DEBUG("No more processes to run");
uintptr_t worker_id =
(uintptr_t) xbt_os_thread_get_specific(sysv_worker_id_key);
// Deduce the initial soul of that body
- next_context = (ParallelUContext*) sysv_workers_context[worker_id];
+ next_context = sysv_workers_context[worker_id];
// When given that soul, the body will wait for the next scheduling round
}
- next_stack = &next_context->uc_;
+ // Will contain the next soul to run, either simulated or initial minion's one
+ ucontext_t* next_stack = &next_context->uc_;
SIMIX_context_set_current(next_context);
// Get that next soul:
#if SIMGRID_HAVE_MC
xbt_assert(mc_model_checker == nullptr, "This must be called from the client");
#endif
-
- smx_actor_t process;
- unsigned int iter;
-
- while (not xbt_dynar_is_empty(simix_global->process_to_run)) {
+ while (not simix_global->process_to_run.empty()) {
SIMIX_process_runall();
- xbt_dynar_foreach(simix_global->process_that_ran, iter, process) {
+ for (smx_actor_t process : simix_global->process_that_ran) {
smx_simcall_t req = &process->simcall;
if (req->call != SIMCALL_NONE && not simgrid::mc::request_is_visible(req))
SIMIX_simcall_handle(req, 0);
if (msg_global==nullptr)
return;
- TRACE_surf_resource_utilization_release();
TRACE_end();
delete msg_global;
msg_global = nullptr;
}
void Mailbox::setReceiver(ActorPtr actor) {
- simix::kernelImmediate([this, actor]() {
- this->pimpl_->setReceiver(actor);
- });
+ simix::kernelImmediate([this, actor]() { this->pimpl_->setReceiver(actor); });
}
/** @brief get the receiver (process associated to the mailbox) */
static xbt_dynar_t result;
static std::unordered_map<std::string, SD_task_t> jobs;
-static xbt_dict_t files;
+static std::unordered_map<std::string, SD_task_t> files;
static SD_task_t current_job;
static SD_task_t root_task;
static SD_task_t end_task;
*/
xbt_dynar_t SD_daxload(const char *filename)
{
- xbt_dict_cursor_t cursor;
SD_task_t file;
- char* name;
FILE* in_file = fopen(filename, "r");
xbt_assert(in_file, "Unable to open \"%s\"\n", filename);
input_buffer = dax__create_buffer(in_file, 10);
dax_lineno = 1;
result = xbt_dynar_new(sizeof(SD_task_t), dax_task_free);
- files = xbt_dict_new_homogeneous(&dax_task_free);
root_task = SD_task_create_comp_seq("root", nullptr, 0);
/* by design the root task is always SCHEDULABLE */
SD_task_set_state(root_task, SD_SCHEDULABLE);
* Files not consumed in the system are said to be consumed by end task (bottom of DAG).
*/
- xbt_dict_foreach(files, cursor, name, file) {
+ for (auto elm : files) {
+ file = elm.second;
SD_task_t newfile;
if (file->predecessors->empty()) {
for (SD_task_t it : *file->successors) {
xbt_dynar_push(result, &end_task);
/* Free previous copy of the files */
- xbt_dict_free(&files);
+ for (auto elm : files)
+ SD_task_destroy(elm.second);
unsigned int cpt;
xbt_dynar_foreach(result, cpt, file) {
if (SD_task_get_kind(file) == SD_TASK_COMM_E2E) {
bool is_input = (A_dax__uses_link == A_dax__uses_link_input);
XBT_DEBUG("See <uses file=%s %s>",A_dax__uses_file,(is_input?"in":"out"));
- SD_task_t file = static_cast<SD_task_t>(xbt_dict_get_or_null(files, A_dax__uses_file));
- if (file == nullptr) {
+ auto it = files.find(A_dax__uses_file);
+ SD_task_t file;
+ if (it == files.end()) {
file = SD_task_create_comm_e2e(A_dax__uses_file, nullptr, size);
sd_global->initial_tasks->erase(file);
- xbt_dict_set(files, A_dax__uses_file, file, nullptr);
+ files[A_dax__uses_file] = file;
} else {
+ file = it->second;
if (file->amount < size || file->amount > size) {
XBT_WARN("Ignore file %s size redefinition from %.0f to %.0f", A_dax__uses_file, SD_task_get_amount(file), size);
}
-/* Copyright (c) 2006-2015. The SimGrid Team.
+/* Copyright (c) 2006-2017. The SimGrid Team.
* All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
class Global {
public:
explicit Global();
+ Global(const Global&) = delete;
~Global();
bool watch_point_reached; /* has a task just reached a watch point? */
std::set<SD_task_t> *initial_tasks;
/* The parameters of network models */
- // real default for "network/sender-gap" is set in network_smpi.cpp:
- sg_sender_gap = NAN;
- simgrid::config::bindFlag(sg_sender_gap, {"network/sender-gap", "network/sender_gap"},
- "Minimum gap between two overlapping sends");
-
sg_latency_factor = 13.01; // comes from the default LV08 network model
simgrid::config::bindFlag(sg_latency_factor, {"network/latency-factor", "network/latency_factor"},
"Correction factor to apply to the provided latency (default value set by network model)");
xbt_os_mutex_acquire(simix_global->mutex);
/* cancel non-blocking communications */
- smx_activity_t synchro = process->comms.front();
while (not process->comms.empty()) {
+ smx_activity_t synchro = process->comms.front();
+ process->comms.pop_front();
simgrid::kernel::activity::CommImplPtr comm =
boost::static_pointer_cast<simgrid::kernel::activity::CommImpl>(synchro);
} else {
xbt_die("Communication synchro %p is in my list but I'm not the sender nor the receiver", synchro.get());
}
- process->comms.pop_front();
- synchro = process->comms.front();
comm->cancel();
}
/* Now insert it in the global process list and in the process to run list */
simix_global->process_list[process->pid] = process;
XBT_DEBUG("Inserting %s(%s) in the to_run list", process->cname(), host->getCname());
- xbt_dynar_push_as(simix_global->process_to_run, smx_actor_t, process);
+ simix_global->process_to_run.push_back(process);
intrusive_ptr_add_ref(process);
/* Tracing the process creation */
/* Now insert it in the global process list and in the process to run list */
simix_global->process_list[process->pid] = process;
XBT_DEBUG("Inserting %s(%s) in the to_run list", process->cname(), host->getCname());
- xbt_dynar_push_as(simix_global->process_to_run, smx_actor_t, process);
+ simix_global->process_to_run.push_back(process);
/* Tracing the process creation */
TRACE_msg_process_create(process->cname(), process->pid, process->host);
{
SIMIX_context_runall();
- xbt_dynar_t tmp = simix_global->process_that_ran;
- simix_global->process_that_ran = simix_global->process_to_run;
- simix_global->process_to_run = tmp;
- xbt_dynar_reset(simix_global->process_to_run);
+ simix_global->process_to_run.swap(simix_global->process_that_ran);
+ simix_global->process_to_run.clear();
}
void simcall_HANDLER_process_kill(smx_simcall_t simcall, smx_actor_t process) {
process->waiting_synchro = nullptr;
}
- if (not xbt_dynar_member(simix_global->process_to_run, &(process)) && process != issuer) {
+ if (std::find(begin(simix_global->process_to_run), end(simix_global->process_to_run), process) ==
+ end(simix_global->process_to_run) &&
+ process != issuer) {
XBT_DEBUG("Inserting %s in the to_run list", process->name.c_str());
- xbt_dynar_push_as(simix_global->process_to_run, smx_actor_t, process);
+ simix_global->process_to_run.push_back(process);
}
}
boost::dynamic_pointer_cast<simgrid::kernel::activity::SleepImpl>(process->waiting_synchro);
if (sleep != nullptr) {
SIMIX_process_sleep_destroy(process->waiting_synchro);
- if (not xbt_dynar_member(simix_global->process_to_run, &(process)) && process != SIMIX_process_self()) {
+ if (std::find(begin(simix_global->process_to_run), end(simix_global->process_to_run), process) ==
+ end(simix_global->process_to_run) &&
+ process != SIMIX_process_self()) {
XBT_DEBUG("Inserting %s in the to_run list", process->name.c_str());
- xbt_dynar_push_as(simix_global->process_to_run, smx_actor_t, process);
+ simix_global->process_to_run.push_back(process);
}
}
}
/** @brief Returns the list of processes to run. */
-xbt_dynar_t SIMIX_process_get_runnable()
+const std::vector<smx_actor_t>& simgrid::simix::process_get_runnable()
{
return simix_global->process_to_run;
}
simcall->issuer->name.c_str(), simcall->issuer);
simcall->issuer->simcall.call = SIMCALL_NONE;
/* This check should be useless and slows everyone. Reactivate if you see something weird in process scheduling. */
- // if (xbt_dynar_member(simix_global->process_to_run, &(simcall->issuer)))
+ // if (std::find(begin(simix_global->process_to_run), end(simix_global->process_to_run), simcall->issuer) !=
+ // end(simix_global->process_to_run))
// DIE_IMPOSSIBLE;
- xbt_dynar_push_as(simix_global->process_to_run, smx_actor_t, simcall->issuer);
+ simix_global->process_to_run.push_back(simcall->issuer);
}
}
#include <csignal> /* Signal handling */
#include <cstdlib>
+#include <xbt/algorithm.hpp>
#include <xbt/functional.hpp>
#include "simgrid/s4u/Engine.hpp"
simix_global = std::unique_ptr<simgrid::simix::Global>(new simgrid::simix::Global());
simgrid::simix::ActorImpl proc;
- simix_global->process_to_run = xbt_dynar_new(sizeof(smx_actor_t), nullptr);
- simix_global->process_that_ran = xbt_dynar_new(sizeof(smx_actor_t), nullptr);
simix_global->process_to_destroy = xbt_swag_new(xbt_swag_offset(proc, destroy_hookup));
simix_global->maestro_process = nullptr;
simix_global->create_process_function = &SIMIX_process_create;
smx_cleaned = 1;
XBT_DEBUG("SIMIX_clean called. Simulation's over.");
- if (not xbt_dynar_is_empty(simix_global->process_to_run) && SIMIX_get_clock() <= 0.0) {
+ if (not simix_global->process_to_run.empty() && SIMIX_get_clock() <= 0.0) {
XBT_CRITICAL(" ");
XBT_CRITICAL("The time is still 0, and you still have processes ready to run.");
XBT_CRITICAL("It seems that you forgot to run the simulation that you setup.");
xbt_heap_free(simix_timers);
simix_timers = nullptr;
/* Free the remaining data structures */
- xbt_dynar_free(&simix_global->process_to_run);
- xbt_dynar_free(&simix_global->process_that_ran);
+ simix_global->process_to_run.clear();
+ simix_global->process_that_ran.clear();
xbt_swag_free(simix_global->process_to_destroy);
simix_global->process_list.clear();
simix_global->process_to_destroy = nullptr;
}
}
-static int process_syscall_color(void *p)
-{
- switch ((*(smx_actor_t *)p)->simcall.call) {
- case SIMCALL_NONE:
- case SIMCALL_PROCESS_KILL:
- return 2;
- // case SIMCALL_PROCESS_RESUME:
- // return 1;
- default:
- return 0;
- }
-}
-
/** Wake up all processes waiting for a Surf action to finish */
static void SIMIX_wake_processes()
{
double time = 0;
do {
- XBT_DEBUG("New Schedule Round; size(queue)=%lu", xbt_dynar_length(simix_global->process_to_run));
+ XBT_DEBUG("New Schedule Round; size(queue)=%zu", simix_global->process_to_run.size());
SIMIX_execute_tasks();
- while (not xbt_dynar_is_empty(simix_global->process_to_run)) {
- XBT_DEBUG("New Sub-Schedule Round; size(queue)=%lu", xbt_dynar_length(simix_global->process_to_run));
+ while (not simix_global->process_to_run.empty()) {
+ XBT_DEBUG("New Sub-Schedule Round; size(queue)=%zu", simix_global->process_to_run.size());
/* Run all processes that are ready to run, possibly in parallel */
SIMIX_process_runall();
/* Move all killer processes to the end of the list, because killing a process that have an ongoing simcall is a bad idea */
- xbt_dynar_three_way_partition(simix_global->process_that_ran, process_syscall_color);
+ simgrid::xbt::three_way_partition(begin(simix_global->process_that_ran), end(simix_global->process_that_ran),
+ [](smx_actor_t p) {
+ switch (p->simcall.call) {
+ case SIMCALL_NONE:
+ case SIMCALL_PROCESS_KILL:
+ return 2;
+ // case SIMCALL_PROCESS_RESUME:
+ // return 1;
+ default:
+ return 0;
+ }
+ });
/* answer sequentially and in a fixed arbitrary order all the simcalls that were issued during that sub-round */
* That would thus be a pure waste of time.
*/
- unsigned int iter;
- smx_actor_t process;
- xbt_dynar_foreach(simix_global->process_that_ran, iter, process) {
+ for (smx_actor_t process : simix_global->process_that_ran) {
if (process->simcall.call != SIMCALL_NONE) {
SIMIX_simcall_handle(&process->simcall, 0);
}
/* Clean processes to destroy */
SIMIX_process_empty_trash();
- XBT_DEBUG("### time %f, #processes %zu, #to_run %lu", time, simix_global->process_list.size(),
- xbt_dynar_length(simix_global->process_to_run));
+ XBT_DEBUG("### time %f, #processes %zu, #to_run %zu", time, simix_global->process_list.size(),
+ simix_global->process_to_run.size());
- if (xbt_dynar_is_empty(simix_global->process_to_run) && not simix_global->process_list.empty())
+ if (simix_global->process_to_run.empty() && not simix_global->process_list.empty())
simgrid::simix::onDeadlock();
- } while (time > -1.0 || not xbt_dynar_is_empty(simix_global->process_to_run));
+ } while (time > -1.0 || not simix_global->process_to_run.empty());
if (simix_global->process_list.size() != 0) {
#include <signal.h>
#include <map>
+#include <vector>
/********************************** Simix Global ******************************/
public:
smx_context_factory_t context_factory = nullptr;
- xbt_dynar_t process_to_run = nullptr;
- xbt_dynar_t process_that_ran = nullptr;
+ std::vector<smx_actor_t> process_to_run;
+ std::vector<smx_actor_t> process_that_ran;
std::map<aid_t, smx_actor_t> process_list;
#if SIMGRID_HAVE_MC
/* MCer cannot read the std::map above in the remote process, so we copy the info it needs in a dynar.
#include <cctype>
#include <cstdarg>
#include <cwchar>
+#include <deque>
#include <simgrid/sg_config.h>
XBT_LOG_NEW_DEFAULT_SUBCATEGORY(instr_smpi, instr, "Tracing SMPI");
-static xbt_dict_t keys;
+static std::unordered_map<char*, std::deque<std::string>*> keys;
static const char *smpi_colors[] ={
"recv", "1 0 0",
//get the dynar for src#dst
char aux[INSTR_DEFAULT_STR_SIZE];
snprintf(aux, INSTR_DEFAULT_STR_SIZE, "%d#%d#%d#%d", src, dst, tag, send);
- xbt_dynar_t d = static_cast<xbt_dynar_t>(xbt_dict_get_or_null(keys, aux));
+ auto it = keys.find(aux);
+ std::deque<std::string>* d;
- if (d == nullptr) {
- d = xbt_dynar_new(sizeof(char *), &xbt_free_ref);
- xbt_dict_set(keys, aux, d, nullptr);
- }
+ if (it == keys.end()) {
+ d = new std::deque<std::string>;
+ keys[aux] = d;
+ } else
+ d = it->second;
//generate the key
static unsigned long long counter = 0;
snprintf(key, n, "%d_%d_%d_%llu", src, dst, tag, counter);
//push it
- char *a = static_cast<char*> (xbt_strdup(key));
- xbt_dynar_push_as(d, char *, a);
+ d->push_back(key);
return key;
}
{
char aux[INSTR_DEFAULT_STR_SIZE];
snprintf(aux, INSTR_DEFAULT_STR_SIZE, "%d#%d#%d#%d", src, dst, tag, send==1?0:1);
- xbt_dynar_t d = static_cast<xbt_dynar_t>(xbt_dict_get_or_null(keys, aux));
-
- // first posted
- if(xbt_dynar_is_empty(d)){
- TRACE_smpi_put_key(src, dst, tag, key, n, send);
- return key;
+ auto it = keys.find(aux);
+ if (it == keys.end()) {
+ // first posted
+ TRACE_smpi_put_key(src, dst, tag, key, n, send);
+ } else {
+ snprintf(key, n, "%s", it->second->front().c_str());
+ it->second->pop_front();
}
-
- char *s = xbt_dynar_get_as (d, 0, char *);
- snprintf (key, n, "%s", s);
- xbt_dynar_remove_at (d, 0, nullptr);
return key;
}
-static xbt_dict_t process_category;
+static std::unordered_map<smx_actor_t, std::string> process_category;
static void cleanup_extra_data (instr_extra_data extra){
if(extra!=nullptr){
//declare category
TRACE_category (category);
- char processid[INSTR_DEFAULT_STR_SIZE];
- snprintf (processid, INSTR_DEFAULT_STR_SIZE, "%p", SIMIX_process_self());
- if (xbt_dict_get_or_null (process_category, processid))
- xbt_dict_remove (process_category, processid);
if (category != nullptr)
- xbt_dict_set (process_category, processid, xbt_strdup(category), nullptr);
+ process_category[SIMIX_process_self()] = category;
}
const char *TRACE_internal_smpi_get_category ()
if (not TRACE_smpi_is_enabled())
return nullptr;
- char processid[INSTR_DEFAULT_STR_SIZE];
- snprintf (processid, INSTR_DEFAULT_STR_SIZE, "%p", SIMIX_process_self());
- return static_cast<char*>(xbt_dict_get_or_null (process_category, processid));
+ auto it = process_category.find(SIMIX_process_self());
+ return (it == process_category.end()) ? nullptr : it->second.c_str();
}
void TRACE_smpi_alloc()
{
- keys = xbt_dict_new_homogeneous(xbt_dynar_free_voidp);
- process_category = xbt_dict_new_homogeneous(xbt_free_f);
+ // for symmetry
}
void TRACE_smpi_release()
{
- xbt_dict_free(&keys);
- xbt_dict_free(&process_category);
+ for (auto elm : keys)
+ delete elm.second;
}
void TRACE_smpi_init(int rank)
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
-#include "src/internal_config.h"
#include "private.h"
#include "private.hpp"
#include "simgrid/modelchecker.h"
-#include "src/mc/mc_replay.h"
-#include "smpi_process.hpp"
#include "smpi_comm.hpp"
+#include "smpi_process.hpp"
+#include "src/internal_config.h"
+#include "src/mc/mc_replay.h"
+#include <unordered_map>
#ifndef WIN32
#include <sys/mman.h>
XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_bench, smpi, "Logging specific to SMPI (benchmarking)");
-
-xbt_dict_t samples = nullptr; /* Allocated on first use */
-
double smpi_cpu_threshold = -1;
double smpi_host_speed;
double smpi_total_benched_time = 0;
smpi_privatization_region_t smpi_privatization_regions;
-void smpi_bench_destroy()
-{
- xbt_dict_free(&samples);
-}
-
extern "C" XBT_PUBLIC(void) smpi_execute_flops_(double *flops);
void smpi_execute_flops_(double *flops)
{
int benching; /* 1: we are benchmarking; 0: we have enough data, no bench anymore */
} local_data_t;
+std::unordered_map<std::string, local_data_t*> samples; /* Allocated on first use */
+
static char *sample_location(int global, const char *file, int line) {
if (global) {
return bprintf("%s:%d", file, line);
smpi_bench_end(); /* Take time from previous, unrelated computation into account */
smpi_process()->set_sampling(1);
- if (samples==nullptr)
- samples = xbt_dict_new_homogeneous(free);
-
- local_data_t *data = static_cast<local_data_t *>(xbt_dict_get_or_null(samples, loc));
- if (data==nullptr) {
+ auto ld = samples.find(loc);
+ local_data_t* data;
+ if (ld == samples.end()) {
xbt_assert(threshold>0 || iters>0,
"You should provide either a positive amount of iterations to bench, or a positive maximal stderr (or both)");
- data = static_cast<local_data_t *>( xbt_new(local_data_t, 1));
+ data = static_cast<local_data_t*>(xbt_new(local_data_t, 1));
data->count = 0;
data->sum = 0.0;
data->sum_pow2 = 0.0;
data->threshold = threshold;
data->benching = 1; // If we have no data, we need at least one
data->mean = 0;
- xbt_dict_set(samples, loc, data, nullptr);
+ samples[loc] = data;
XBT_DEBUG("XXXXX First time ever on benched nest %s.",loc);
} else {
+ data = ld->second;
if (data->iters != iters || data->threshold != threshold) {
XBT_ERROR("Asked to bench block %s with different settings %d, %f is not %d, %f. "
"How did you manage to give two numbers at the same line??",
char *loc = sample_location(global, file, line);
int res;
- xbt_assert(samples, "Y U NO use SMPI_SAMPLE_* macros? Stop messing directly with smpi_sample_* functions!");
- local_data_t *data = static_cast<local_data_t *>(xbt_dict_get(samples, loc));
+ xbt_assert(not samples.empty(),
+ "Y U NO use SMPI_SAMPLE_* macros? Stop messing directly with smpi_sample_* functions!");
+ local_data_t* data = samples.at(loc);
XBT_DEBUG("sample2 %s",loc);
xbt_free(loc);
{
char *loc = sample_location(global, file, line);
- xbt_assert(samples, "Y U NO use SMPI_SAMPLE_* macros? Stop messing directly with smpi_sample_* functions!");
- local_data_t *data = static_cast<local_data_t *>(xbt_dict_get(samples, loc));
+ xbt_assert(not samples.empty(),
+ "Y U NO use SMPI_SAMPLE_* macros? Stop messing directly with smpi_sample_* functions!");
+ local_data_t* data = samples.at(loc);
XBT_DEBUG("sample3 %s",loc);
xbt_free(loc);
}
extern "C" { /** These functions will be called from the user code **/
- smpi_trace_call_location_t* smpi_trace_get_call_location() {
- return smpi_process()->call_location();
- }
+smpi_trace_call_location_t* smpi_trace_get_call_location()
+{
+ return smpi_process()->call_location();
+}
- void smpi_trace_set_call_location(const char* file, const int line) {
- smpi_trace_call_location_t* loc = smpi_process()->call_location();
+void smpi_trace_set_call_location(const char* file, const int line)
+{
+ smpi_trace_call_location_t* loc = smpi_process()->call_location();
- loc->previous_filename = loc->filename;
- loc->previous_linenumber = loc->linenumber;
- loc->filename = file;
- loc->linenumber = line;
- }
+ loc->previous_filename = loc->filename;
+ loc->previous_linenumber = loc->linenumber;
+ loc->filename = file;
+ loc->linenumber = line;
+}
- /**
- * Required for Fortran bindings
- */
- void smpi_trace_set_call_location_(const char* file, int* line) {
- smpi_trace_set_call_location(file, *line);
- }
+/** Required for Fortran bindings */
+void smpi_trace_set_call_location_(const char* file, int* line)
+{
+ smpi_trace_set_call_location(file, *line);
+}
- /**
- * Required for Fortran if -fsecond-underscore is activated
- */
- void smpi_trace_set_call_location__(const char* file, int* line) {
- smpi_trace_set_call_location(file, *line);
- }
+/** Required for Fortran if -fsecond-underscore is activated */
+void smpi_trace_set_call_location__(const char* file, int* line)
+{
+ smpi_trace_set_call_location(file, *line);
+}
+}
+
+void smpi_bench_destroy()
+{
+ for (auto elm : samples)
+ xbt_free(elm.second);
}
* Associated data and metadata are used as follows:
*
* mmap #1
- * `allocs' dict ---- -.
+ * `allocs' map ---- -.
* ---------- shared_data_t shared_metadata_t / | | |
* .->| <name> | ---> -------------------- <--. ----------------- | | | |
* | ---------- | fd of <name> | | | size of mmap | --| | | |
* `----------------- | <name> | | ----------------- ---- |
* -------------------- | ^ |
* | | |
- * | | `allocs_metadata' dict |
+ * | | `allocs_metadata' map |
* | | ---------------------- |
* | `-- | <addr of mmap #1> |<-'
* | .-- | <addr of mmap #2> |<-.
#include "private.h"
#include "private.hpp"
-#include "xbt/dict.h"
-#include "xbt/ex.hpp"
#include <cerrno>
#include <sys/types.h>
} shared_metadata_t;
std::map<void*, shared_metadata_t> allocs_metadata;
-xbt_dict_t calls = nullptr; /* Allocated on first use */
+std::map<std::string, void*> calls;
+
#ifndef WIN32
static int smpi_shared_malloc_bogusfile = -1;
static int smpi_shared_malloc_bogusfile_huge_page = -1;
{
allocs.clear();
allocs_metadata.clear();
- xbt_dict_free(&calls);
+ calls.clear();
}
static size_t shm_size(int fd) {
int smpi_shared_known_call(const char* func, const char* input)
{
- char* loc = bprintf("%s:%s", func, input);
- int known = 0;
-
- if (calls==nullptr) {
- calls = xbt_dict_new_homogeneous(nullptr);
- }
- try {
- xbt_dict_get(calls, loc); /* Succeed or throw */
- known = 1;
- xbt_free(loc);
- }
- catch (xbt_ex& ex) {
- xbt_free(loc);
- if (ex.category != not_found_error)
- throw;
- }
- catch(...) {
- xbt_free(loc);
- throw;
- }
- return known;
+ std::string loc = std::string(func) + ":" + input;
+ return calls.find(loc) != calls.end();
}
void* smpi_shared_get_call(const char* func, const char* input) {
- char* loc = bprintf("%s:%s", func, input);
+ std::string loc = std::string(func) + ":" + input;
- if (calls == nullptr)
- calls = xbt_dict_new_homogeneous(nullptr);
- void* data = xbt_dict_get(calls, loc);
- xbt_free(loc);
- return data;
+ return calls.at(loc);
}
void* smpi_shared_set_call(const char* func, const char* input, void* data) {
- char* loc = bprintf("%s:%s", func, input);
-
- if (calls == nullptr)
- calls = xbt_dict_new_homogeneous(nullptr);
- xbt_dict_set(calls, loc, data, nullptr);
- xbt_free(loc);
+ std::string loc = std::string(func) + ":" + input;
+ calls[loc] = data;
return data;
}
-
/**
* @brief Get the content of the current Storage
*
- * @return A xbt_dict_t with path as keys and size in bytes as values
+ * @return A map with path as keys and size in bytes as values
*/
virtual std::map<std::string, sg_size_t>* getContent();
}
//register EDGE types for triva configuration
- xbt_dict_set (trivaEdgeTypes, link_type->name, xbt_strdup("1"), nullptr);
+ trivaEdgeTypes.insert(link_type->name);
//create the link
static long long counter = 0;
XBT_LOG_NEW_DEFAULT_SUBCATEGORY(instr_surf, instr, "Tracing Surf");
-void TRACE_surf_alloc()
-{
- TRACE_surf_resource_utilization_alloc();
-}
-
-void TRACE_surf_release()
-{
- TRACE_surf_resource_utilization_release();
-}
-
void TRACE_surf_host_set_speed(double date, const char *resource, double speed)
{
if (TRACE_categorized() || TRACE_uncategorized() || TRACE_platform()) {
XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(surf_network);
-double sg_sender_gap = 0.0;
double sg_latency_factor = 1.0; /* default value; can be set by model or from command line */
double sg_bandwidth_factor = 1.0; /* default value; can be set by model or from command line */
double sg_weight_S_parameter = 0.0; /* default value; can be set by model or from command line */
action->latCurrent_ = action->latency_;
action->latency_ *= latencyFactor(size);
action->rate_ = bandwidthConstraint(action->rate_, bandwidth_bound, size);
- if (haveGap_) {
- xbt_assert(not route->empty(),
- "Using a model with a gap (e.g., SMPI) with a platform without links (e.g. vivaldi)!!!");
-
- gapAppend(size, route->at(0), action);
- XBT_DEBUG("Comm %p: %s -> %s gap=%f (lat=%f)", action, src->getCname(), dst->getCname(), action->senderGap_,
- action->latency_);
- }
int constraints_per_variable = route->size();
if (back_route != nullptr)
return action;
}
-void NetworkCm02Model::gapAppend(double size, const LinkImpl* link, NetworkAction* action){
- // Nothing
-};
-
/************
* Resource *
************/
lastValue_ = lmm_variable_getvalue(getVariable());
}
-void NetworkCm02Link::gapAppend(double size, const LinkImpl* link, NetworkAction* action){
- // Nothing
-};
}
}
void updateActionsStateLazy(double now, double delta) override;
void updateActionsStateFull(double now, double delta) override;
Action* communicate(s4u::Host* src, s4u::Host* dst, double size, double rate) override;
- virtual void gapAppend(double size, const LinkImpl* link, NetworkAction* action);
-
-protected:
- bool haveGap_ = false;
};
/************
void apply_event(tmgr_trace_event_t event, double value) override;
void setBandwidth(double value) override;
void setLatency(double value) override;
- virtual void gapAppend(double size, const LinkImpl* link, NetworkAction* action);
};
/**********
NetworkCm02Action(Model* model, double cost, bool failed) : NetworkAction(model, cost, failed){};
virtual ~NetworkCm02Action() = default;
void updateRemainingLazy(double now) override;
-
-protected:
- double senderGap_;
};
}
}
NetworkIBModel::NetworkIBModel() : NetworkSmpiModel()
{
- haveGap_ = false;
const char* IB_factors_string = xbt_cfg_get_string("smpi/IB-penalty-factors");
std::vector<std::string> radical_elements;
boost::split(radical_elements, IB_factors_string, boost::is_any_of(";"));
-/* Copyright (c) 2013-2015. The SimGrid Team.
+/* Copyright (c) 2013-2017. The SimGrid Team.
* All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
std::vector<s_smpi_factor_t> smpi_bw_factor;
std::vector<s_smpi_factor_t> smpi_lat_factor;
-xbt_dict_t gap_lookup = nullptr;
-
/*********
* Model *
*********/
surf_network_model = new simgrid::surf::NetworkSmpiModel();
all_existing_models->push_back(surf_network_model);
- xbt_cfg_setdefault_double("network/sender-gap", 10e-6);
xbt_cfg_setdefault_double("network/weight-S", 8775);
}
namespace simgrid {
- namespace surf {
-
- NetworkSmpiModel::NetworkSmpiModel() : NetworkCm02Model()
- {
- haveGap_ = true;
- }
-
- NetworkSmpiModel::~NetworkSmpiModel()
- {
- xbt_dict_free(&gap_lookup);
- }
-
- double NetworkSmpiModel::bandwidthFactor(double size)
- {
- if (smpi_bw_factor.empty())
- smpi_bw_factor = parse_factor(xbt_cfg_get_string("smpi/bw-factor"));
-
- double current = 1.0;
- for (const auto& fact : smpi_bw_factor) {
- if (size <= fact.factor) {
- XBT_DEBUG("%f <= %zu return %f", size, fact.factor, current);
- return current;
- } else
- current = fact.values.front();
- }
- XBT_DEBUG("%f > %zu return %f", size, smpi_bw_factor.back().factor, current);
+namespace surf {
- return current;
- }
-
- double NetworkSmpiModel::latencyFactor(double size)
- {
- if (smpi_lat_factor.empty())
- smpi_lat_factor = parse_factor(xbt_cfg_get_string("smpi/lat-factor"));
-
- double current=1.0;
- for (const auto& fact : smpi_lat_factor) {
- if (size <= fact.factor) {
- XBT_DEBUG("%f <= %zu return %f", size, fact.factor, current);
- return current;
- }else
- current=fact.values.front();
- }
- XBT_DEBUG("%f > %zu return %f", size, smpi_lat_factor.back().factor, current);
+NetworkSmpiModel::NetworkSmpiModel() : NetworkCm02Model()
+{
+}
+
+NetworkSmpiModel::~NetworkSmpiModel() = default;
+double NetworkSmpiModel::bandwidthFactor(double size)
+{
+ if (smpi_bw_factor.empty())
+ smpi_bw_factor = parse_factor(xbt_cfg_get_string("smpi/bw-factor"));
+
+ double current = 1.0;
+ for (const auto& fact : smpi_bw_factor) {
+ if (size <= fact.factor) {
+ XBT_DEBUG("%f <= %zu return %f", size, fact.factor, current);
return current;
- }
+ } else
+ current = fact.values.front();
+ }
+ XBT_DEBUG("%f > %zu return %f", size, smpi_bw_factor.back().factor, current);
+
+ return current;
+}
- double NetworkSmpiModel::bandwidthConstraint(double rate, double bound, double size)
- {
- return rate < 0 ? bound : std::min(bound, rate * bandwidthFactor(size));
- }
+double NetworkSmpiModel::latencyFactor(double size)
+{
+ if (smpi_lat_factor.empty())
+ smpi_lat_factor = parse_factor(xbt_cfg_get_string("smpi/lat-factor"));
- /************
- * Resource *
- ************/
+ double current = 1.0;
+ for (const auto& fact : smpi_lat_factor) {
+ if (size <= fact.factor) {
+ XBT_DEBUG("%f <= %zu return %f", size, fact.factor, current);
+ return current;
+ } else
+ current = fact.values.front();
+ }
+ XBT_DEBUG("%f > %zu return %f", size, smpi_lat_factor.back().factor, current);
+ return current;
+}
+double NetworkSmpiModel::bandwidthConstraint(double rate, double bound, double size)
+{
+ return rate < 0 ? bound : std::min(bound, rate * bandwidthFactor(size));
+}
- /**********
- * Action *
- **********/
+/************
+ * Resource *
+ ************/
- }
+/**********
+ * Action *
+ **********/
+}
}
-/* Copyright (c) 2013-2015. The SimGrid Team.
+/* Copyright (c) 2013-2017. The SimGrid Team.
* All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
NetworkSmpiModel();
~NetworkSmpiModel();
- using NetworkCm02Model::gapAppend; // Explicit about overloaded method (silence Woverloaded-virtual from clang)
double latencyFactor(double size);
double bandwidthFactor(double size);
double bandwidthConstraint(double rate, double bound, double size);
if (not future_evt_set)
future_evt_set = new simgrid::trace_mgr::future_evt_set();
- TRACE_surf_alloc();
- simgrid::surf::surfExitCallbacks.connect(TRACE_surf_release);
-
sg_config_init(argc, argv);
if (MC_is_active())
-/* Copyright (c) 2004-2016. The SimGrid Team. All rights reserved. */
+/* Copyright (c) 2004-2017. The SimGrid Team. All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
/* user-visible parameters */
extern XBT_PRIVATE double sg_tcp_gamma;
-extern XBT_PRIVATE double sg_sender_gap;
extern XBT_PRIVATE double sg_latency_factor;
extern XBT_PRIVATE double sg_bandwidth_factor;
extern XBT_PRIVATE double sg_weight_S_parameter;
*/
XBT_PUBLIC_DATA(std::vector<sg_host_t>) host_that_restart;
-
-extern XBT_PRIVATE double sg_sender_gap;
-
namespace simgrid {
namespace surf {
#ifndef SURF_SURFXML_PARSE_H
#define SURF_SURFXML_PARSE_H
-#include <xbt/dict.h>
#include <xbt/function_types.h>
#include <xbt/misc.h>
#include <xbt/signal.hpp>
#include "src/instr/instr_private.h" // TRACE_start(). FIXME: remove by subscribing tracing to the surf signals
#include "src/surf/cpu_interface.hpp"
#include "src/surf/network_interface.hpp"
-#include "xbt/dict.h"
#include "xbt/log.h"
#include "xbt/misc.h"
#include "xbt/str.h"
XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(surf_parse);
-
/* Trace related stuff */
XBT_PRIVATE std::unordered_map<std::string, tmgr_trace_t> traces_set_list;
XBT_PRIVATE std::unordered_map<std::string, std::string> trace_connect_list_host_avail;
std::map<std::string, std::string>* current_model_property_set = nullptr;
int ZONE_TAG = 0; // Whether we just opened a zone tag (to see what to do with the properties)
-/* dictionary of random generator data */
-xbt_dict_t random_data_list = nullptr;
-
YY_BUFFER_STATE surf_input_buffer;
FILE *surf_file_to_parse = nullptr;
/* a generic DYNamic ARray implementation. */
-/* Copyright (c) 2004-2015. The SimGrid Team.
+/* Copyright (c) 2004-2017. The SimGrid Team.
* All rights reserved. */
/* This program is free software; you can redistribute it and/or modify it
return dynar; // to enable functional uses
}
-/** @brief Sorts a dynar according to their color assuming elements can have only three colors.
- * Since there are only three colors, it is linear and much faster than a classical sort.
- * See for example http://en.wikipedia.org/wiki/Dutch_national_flag_problem
- *
- * \param dynar the dynar to sort
- * \param color the color function of type (int (compar_fn*) (void*) (void*)). The return value of color is assumed to
- * be 0, 1, or 2.
- *
- * At the end of the call, elements with color 0 are at the beginning of the dynar, elements with color 2 are at the
- * end and elements with color 1 are in the middle.
- *
- * Remark: if the elements stored in the dynar are structures, the color function has to retrieve the field to sort
- * first.
- */
-extern "C" void xbt_dynar_three_way_partition(xbt_dynar_t const dynar, int_f_pvoid_t color)
-{
- unsigned long int i;
- unsigned long int p = -1;
- unsigned long int q = dynar->used;
- const unsigned long elmsize = dynar->elmsize;
- char* tmp[elmsize];
- void *elm;
-
- for (i = 0; i < q;) {
- void *elmi = _xbt_dynar_elm(dynar, i);
- int colori = color(elmi);
-
- if (colori == 1) {
- ++i;
- } else {
- if (colori == 0) {
- ++p;
- elm = _xbt_dynar_elm(dynar, p);
- ++i;
- } else { /* colori == 2 */
- --q;
- elm = _xbt_dynar_elm(dynar, q);
- }
- if (elm != elmi) {
- memcpy(tmp, elm, elmsize);
- memcpy(elm, elmi, elmsize);
- memcpy(elmi, tmp, elmsize);
- }
- }
- }
-}
-
/** @brief Transform a dynar into a nullptr terminated array.
*
* \param dynar the dynar to transform
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
-#include <atomic>
-#include <climits>
-
-#include "src/internal_config.h"
-#if HAVE_UNISTD_H
-#include <unistd.h>
-#endif
-
-#ifndef _WIN32
-#include <sys/syscall.h>
-#endif
-
-#if HAVE_FUTEX_H
-#include <climits>
-#include <linux/futex.h>
-#endif
-
-#include "xbt/parmap.h"
#include "xbt/log.h"
-#include "xbt/function_types.h"
-#include "xbt/dynar.h"
-#include "xbt/xbt_os_thread.h"
-#include "xbt/sysdep.h"
-#include "src/simix/smx_private.h"
XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_parmap, xbt, "parmap: parallel map");
-
-typedef enum {
- XBT_PARMAP_WORK,
- XBT_PARMAP_DESTROY
-} e_xbt_parmap_flag_t;
-
-static void xbt_parmap_set_mode(xbt_parmap_t parmap, e_xbt_parmap_mode_t mode);
-static void *xbt_parmap_worker_main(void *parmap);
-static void xbt_parmap_work(xbt_parmap_t parmap);
-
-static void xbt_parmap_posix_master_wait(xbt_parmap_t parmap);
-static void xbt_parmap_posix_worker_signal(xbt_parmap_t parmap);
-static void xbt_parmap_posix_master_signal(xbt_parmap_t parmap);
-static void xbt_parmap_posix_worker_wait(xbt_parmap_t parmap, unsigned round);
-
-#if HAVE_FUTEX_H
-static void xbt_parmap_futex_master_wait(xbt_parmap_t parmap);
-static void xbt_parmap_futex_worker_signal(xbt_parmap_t parmap);
-static void xbt_parmap_futex_master_signal(xbt_parmap_t parmap);
-static void xbt_parmap_futex_worker_wait(xbt_parmap_t parmap, unsigned round);
-static void futex_wait(unsigned *uaddr, unsigned val);
-static void futex_wake(unsigned *uaddr, unsigned val);
-#endif
-
-static void xbt_parmap_busy_master_wait(xbt_parmap_t parmap);
-static void xbt_parmap_busy_worker_signal(xbt_parmap_t parmap);
-static void xbt_parmap_busy_master_signal(xbt_parmap_t parmap);
-static void xbt_parmap_busy_worker_wait(xbt_parmap_t parmap, unsigned round);
-
-/**
- * \brief Parallel map structure
- */
-typedef struct s_xbt_parmap {
- e_xbt_parmap_flag_t status; /**< is the parmap active or being destroyed? */
- unsigned work; /**< index of the current round */
- unsigned thread_counter; /**< number of workers that have done the work */
-
- unsigned int num_workers; /**< total number of worker threads including the controller */
- xbt_os_thread_t *workers; /**< worker thread handlers */
- void_f_pvoid_t fun; /**< function to run in parallel on each element of data */
- xbt_dynar_t data; /**< parameters to pass to fun in parallel */
- std::atomic<unsigned int> index; /**< index of the next element of data to pick */
-
- /* posix only */
- xbt_os_cond_t ready_cond;
- xbt_os_mutex_t ready_mutex;
- xbt_os_cond_t done_cond;
- xbt_os_mutex_t done_mutex;
-
- /* fields that depend on the synchronization mode */
- e_xbt_parmap_mode_t mode; /**< synchronization mode */
- void (*master_wait_f)(xbt_parmap_t); /**< wait for the workers to have done the work */
- void (*worker_signal_f)(xbt_parmap_t); /**< signal the master that a worker has done the work */
- void (*master_signal_f)(xbt_parmap_t); /**< wakes the workers threads to process tasks */
- void (*worker_wait_f)(xbt_parmap_t, unsigned); /**< waits for more work */
-} s_xbt_parmap_t;
-
-/**
- * \brief Thread data transmission structure
- */
-typedef struct s_xbt_parmap_thread_data{
- xbt_parmap_t parmap;
- int worker_id;
-} s_xbt_parmap_thread_data_t;
-
-typedef s_xbt_parmap_thread_data_t *xbt_parmap_thread_data_t;
-
-/**
- * \brief Creates a parallel map object
- * \param num_workers number of worker threads to create
- * \param mode how to synchronize the worker threads
- * \return the parmap created
- */
-xbt_parmap_t xbt_parmap_new(unsigned int num_workers, e_xbt_parmap_mode_t mode)
-{
- XBT_DEBUG("Create new parmap (%u workers)", num_workers);
-
- /* Initialize the thread pool data structure */
- xbt_parmap_t parmap = new s_xbt_parmap_t();
- parmap->workers = xbt_new(xbt_os_thread_t, num_workers);
-
- parmap->num_workers = num_workers;
- parmap->status = XBT_PARMAP_WORK;
- xbt_parmap_set_mode(parmap, mode);
-
- /* Create the pool of worker threads */
- parmap->workers[0] = nullptr;
-#if HAVE_PTHREAD_SETAFFINITY
- int core_bind = 0;
-#endif
- for (unsigned int i = 1; i < num_workers; i++) {
- xbt_parmap_thread_data_t data = xbt_new0(s_xbt_parmap_thread_data_t, 1);
- data->parmap = parmap;
- data->worker_id = i;
- parmap->workers[i] = xbt_os_thread_create(nullptr, xbt_parmap_worker_main, data, nullptr);
-#if HAVE_PTHREAD_SETAFFINITY
- xbt_os_thread_bind(parmap->workers[i], core_bind);
- if (core_bind != xbt_os_get_numcores() - 1)
- core_bind++;
- else
- core_bind = 0;
-#endif
- }
- return parmap;
-}
-
-/**
- * \brief Destroys a parmap
- * \param parmap the parmap to destroy
- */
-void xbt_parmap_destroy(xbt_parmap_t parmap)
-{
- if (not parmap) {
- return;
- }
-
- parmap->status = XBT_PARMAP_DESTROY;
- parmap->master_signal_f(parmap);
-
- unsigned int i;
- for (i = 1; i < parmap->num_workers; i++)
- xbt_os_thread_join(parmap->workers[i], nullptr);
-
- xbt_os_cond_destroy(parmap->ready_cond);
- xbt_os_mutex_destroy(parmap->ready_mutex);
- xbt_os_cond_destroy(parmap->done_cond);
- xbt_os_mutex_destroy(parmap->done_mutex);
-
- xbt_free(parmap->workers);
- delete parmap;
-}
-
-/**
- * \brief Sets the synchronization mode of a parmap.
- * \param parmap a parallel map object
- * \param mode the synchronization mode
- */
-static void xbt_parmap_set_mode(xbt_parmap_t parmap, e_xbt_parmap_mode_t mode)
-{
- if (mode == XBT_PARMAP_DEFAULT) {
-#if HAVE_FUTEX_H
- mode = XBT_PARMAP_FUTEX;
-#else
- mode = XBT_PARMAP_POSIX;
-#endif
- }
- parmap->mode = mode;
-
- switch (mode) {
- case XBT_PARMAP_POSIX:
- parmap->master_wait_f = &xbt_parmap_posix_master_wait;
- parmap->worker_signal_f = &xbt_parmap_posix_worker_signal;
- parmap->master_signal_f = &xbt_parmap_posix_master_signal;
- parmap->worker_wait_f = &xbt_parmap_posix_worker_wait;
-
- parmap->ready_cond = xbt_os_cond_init();
- parmap->ready_mutex = xbt_os_mutex_init();
- parmap->done_cond = xbt_os_cond_init();
- parmap->done_mutex = xbt_os_mutex_init();
- break;
- case XBT_PARMAP_FUTEX:
-#if HAVE_FUTEX_H
- parmap->master_wait_f = &xbt_parmap_futex_master_wait;
- parmap->worker_signal_f = &xbt_parmap_futex_worker_signal;
- parmap->master_signal_f = &xbt_parmap_futex_master_signal;
- parmap->worker_wait_f = &xbt_parmap_futex_worker_wait;
-
- xbt_os_cond_destroy(parmap->ready_cond);
- xbt_os_mutex_destroy(parmap->ready_mutex);
- xbt_os_cond_destroy(parmap->done_cond);
- xbt_os_mutex_destroy(parmap->done_mutex);
- break;
-#else
- xbt_die("Futex is not available on this OS.");
-#endif
- case XBT_PARMAP_BUSY_WAIT:
- parmap->master_wait_f = &xbt_parmap_busy_master_wait;
- parmap->worker_signal_f = &xbt_parmap_busy_worker_signal;
- parmap->master_signal_f = &xbt_parmap_busy_master_signal;
- parmap->worker_wait_f = &xbt_parmap_busy_worker_wait;
-
- xbt_os_cond_destroy(parmap->ready_cond);
- xbt_os_mutex_destroy(parmap->ready_mutex);
- xbt_os_cond_destroy(parmap->done_cond);
- xbt_os_mutex_destroy(parmap->done_mutex);
- break;
- case XBT_PARMAP_DEFAULT:
- THROW_IMPOSSIBLE;
- break;
- default:
- THROW_IMPOSSIBLE;
- }
-}
-
-/**
- * \brief Applies a list of tasks in parallel.
- * \param parmap a parallel map object
- * \param fun the function to call in parallel
- * \param data each element of this dynar will be passed as an argument to fun
- */
-void xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data)
-{
- /* Assign resources to worker threads (we are maestro here)*/
- parmap->fun = fun;
- parmap->data = data;
- parmap->index = 0;
- parmap->master_signal_f(parmap); // maestro runs futex_wait to wake all the minions (the working threads)
- xbt_parmap_work(parmap); // maestro works with its minions
- parmap->master_wait_f(parmap); // When there is no more work to do, then maestro waits for the last minion to stop
- XBT_DEBUG("Job done"); // ... and proceeds
-}
-
-/**
- * \brief Returns a next task to process.
- *
- * Worker threads call this function to get more work.
- *
- * \return the next task to process, or nullptr if there is no more work
- */
-void* xbt_parmap_next(xbt_parmap_t parmap)
-{
- unsigned int index = parmap->index++;
- if (index < xbt_dynar_length(parmap->data)) {
- return xbt_dynar_get_as(parmap->data, index, void*);
- }
- return nullptr;
-}
-
-static void xbt_parmap_work(xbt_parmap_t parmap)
-{
- unsigned int index = parmap->index++;
- while (index < xbt_dynar_length(parmap->data)){
- parmap->fun(xbt_dynar_get_as(parmap->data, index, void*));
- index = parmap->index++;
- }
-}
-
-/**
- * \brief Main function of a worker thread.
- * \param arg the parmap
- */
-static void *xbt_parmap_worker_main(void *arg)
-{
- xbt_parmap_thread_data_t data = static_cast<xbt_parmap_thread_data_t>(arg);
- xbt_parmap_t parmap = data->parmap;
- unsigned round = 0;
- smx_context_t context = SIMIX_context_new(std::function<void()>(), nullptr, nullptr);
- SIMIX_context_set_current(context);
-
- XBT_DEBUG("New worker thread created");
-
- /* Worker's main loop */
- while (1) {
- round++;
- parmap->worker_wait_f(parmap, round);
- if (parmap->status == XBT_PARMAP_WORK) {
- XBT_DEBUG("Worker %d got a job", data->worker_id);
-
- xbt_parmap_work(parmap);
- parmap->worker_signal_f(parmap);
-
- XBT_DEBUG("Worker %d has finished", data->worker_id);
- /* We are destroying the parmap */
- } else {
- delete context;
- xbt_free(data);
- return nullptr;
- }
- }
-}
-
-#if HAVE_FUTEX_H
-static void futex_wait(unsigned *uaddr, unsigned val)
-{
- XBT_VERB("Waiting on futex %p", uaddr);
- syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, nullptr, nullptr, 0);
-}
-
-static void futex_wake(unsigned *uaddr, unsigned val)
-{
- XBT_VERB("Waking futex %p", uaddr);
- syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, nullptr, nullptr, 0);
-}
-#endif
-
-/**
- * \brief Starts the parmap: waits for all workers to be ready and returns.
- *
- * This function is called by the controller thread.
- *
- * \param parmap a parmap
- */
-static void xbt_parmap_posix_master_wait(xbt_parmap_t parmap)
-{
- xbt_os_mutex_acquire(parmap->done_mutex);
- if (parmap->thread_counter < parmap->num_workers) {
- /* wait for all workers to be ready */
- xbt_os_cond_wait(parmap->done_cond, parmap->done_mutex);
- }
- xbt_os_mutex_release(parmap->done_mutex);
-}
-
-/**
- * \brief Ends the parmap: wakes the controller thread when all workers terminate.
- *
- * This function is called by all worker threads when they end (not including the controller).
- *
- * \param parmap a parmap
- */
-static void xbt_parmap_posix_worker_signal(xbt_parmap_t parmap)
-{
- xbt_os_mutex_acquire(parmap->done_mutex);
- parmap->thread_counter++;
- if (parmap->thread_counter == parmap->num_workers) {
- /* all workers have finished, wake the controller */
- xbt_os_cond_signal(parmap->done_cond);
- }
- xbt_os_mutex_release(parmap->done_mutex);
-}
-
-/**
- * \brief Wakes all workers and waits for them to finish the tasks.
- *
- * This function is called by the controller thread.
- *
- * \param parmap a parmap
- */
-static void xbt_parmap_posix_master_signal(xbt_parmap_t parmap)
-{
- xbt_os_mutex_acquire(parmap->ready_mutex);
- parmap->thread_counter = 1;
- parmap->work++;
- /* wake all workers */
- xbt_os_cond_broadcast(parmap->ready_cond);
- xbt_os_mutex_release(parmap->ready_mutex);
-}
-
-/**
- * \brief Waits for some work to process.
- *
- * This function is called by each worker thread (not including the controller) when it has no more work to do.
- *
- * \param parmap a parmap
- * \param round the expected round number
- */
-static void xbt_parmap_posix_worker_wait(xbt_parmap_t parmap, unsigned round)
-{
- xbt_os_mutex_acquire(parmap->ready_mutex);
- /* wait for more work */
- if (parmap->work != round) {
- xbt_os_cond_wait(parmap->ready_cond, parmap->ready_mutex);
- }
- xbt_os_mutex_release(parmap->ready_mutex);
-}
-
-#if HAVE_FUTEX_H
-/**
- * \brief Starts the parmap: waits for all workers to be ready and returns.
- *
- * This function is called by the controller thread.
- *
- * \param parmap a parmap
- */
-static void xbt_parmap_futex_master_wait(xbt_parmap_t parmap)
-{
- unsigned count = parmap->thread_counter;
- while (count < parmap->num_workers) {
- /* wait for all workers to be ready */
- futex_wait(&parmap->thread_counter, count);
- count = parmap->thread_counter;
- }
-}
-
-/**
- * \brief Ends the parmap: wakes the controller thread when all workers terminate.
- *
- * This function is called by all worker threads when they end (not including the controller).
- *
- * \param parmap a parmap
- */
-static void xbt_parmap_futex_worker_signal(xbt_parmap_t parmap)
-{
- unsigned count = __sync_add_and_fetch(&parmap->thread_counter, 1);
- if (count == parmap->num_workers) {
- /* all workers have finished, wake the controller */
- futex_wake(&parmap->thread_counter, INT_MAX);
- }
-}
-
-/**
- * \brief Wakes all workers and waits for them to finish the tasks.
- *
- * This function is called by the controller thread.
- *
- * \param parmap a parmap
- */
-static void xbt_parmap_futex_master_signal(xbt_parmap_t parmap)
-{
- parmap->thread_counter = 1;
- __sync_add_and_fetch(&parmap->work, 1);
- /* wake all workers */
- futex_wake(&parmap->work, INT_MAX);
-}
-
-/**
- * \brief Waits for some work to process.
- *
- * This function is called by each worker thread (not including the controller) when it has no more work to do.
- *
- * \param parmap a parmap
- * \param round the expected round number
- */
-static void xbt_parmap_futex_worker_wait(xbt_parmap_t parmap, unsigned round)
-{
- unsigned work = parmap->work;
- /* wait for more work */
- while (work != round) {
- futex_wait(&parmap->work, work);
- work = parmap->work;
- }
-}
-#endif
-
-/**
- * \brief Starts the parmap: waits for all workers to be ready and returns.
- *
- * This function is called by the controller thread.
- *
- * \param parmap a parmap
- */
-static void xbt_parmap_busy_master_wait(xbt_parmap_t parmap)
-{
- while (parmap->thread_counter < parmap->num_workers) {
- xbt_os_thread_yield();
- }
-}
-
-/**
- * \brief Ends the parmap: wakes the controller thread when all workers terminate.
- *
- * This function is called by all worker threads when they end.
- *
- * \param parmap a parmap
- */
-static void xbt_parmap_busy_worker_signal(xbt_parmap_t parmap)
-{
- __sync_add_and_fetch(&parmap->thread_counter, 1);
-}
-
-/**
- * \brief Wakes all workers and waits for them to finish the tasks.
- *
- * This function is called by the controller thread.
- *
- * \param parmap a parmap
- */
-static void xbt_parmap_busy_master_signal(xbt_parmap_t parmap)
-{
- parmap->thread_counter = 1;
- __sync_add_and_fetch(&parmap->work, 1);
-}
-
-/**
- * \brief Waits for some work to process.
- *
- * This function is called by each worker thread (not including the controller) when it has no more work to do.
- *
- * \param parmap a parmap
- * \param round the expected round number
- */
-static void xbt_parmap_busy_worker_wait(xbt_parmap_t parmap, unsigned round)
-{
- /* wait for more work */
- while (parmap->work != round) {
- xbt_os_thread_yield();
- }
-}
fs = new std::ifstream(filename, std::ifstream::in);
xbt_assert(fs->is_open(), "Cannot read replay file '%s'", filename);
}
+ ReplayReader(const ReplayReader&) = delete;
~ReplayReader()
{
delete fs;
int lexical_block_variable = 50;
test_local_variable(process.binary_info.get(), "main", "lexical_block_variable", &lexical_block_variable, &cursor);
- s_foo my_foo;
+ s_foo my_foo = {0};
test_type_by_name(process, my_foo);
_exit(0);
set(teshsuite_src ${teshsuite_src} ${CMAKE_CURRENT_SOURCE_DIR}/${x}/${x}.cpp)
endforeach()
+add_executable (bittorrent app-bittorrent/bittorrent.c app-bittorrent/messages.c app-bittorrent/peer.c app-bittorrent/tracker.c app-bittorrent/connection.c)
+target_link_libraries(bittorrent simgrid)
+set_target_properties(bittorrent PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/app-bittorrent)
+foreach (file bittorrent connection messages peer tracker)
+ set(teshsuite_src ${teshsuite_src} ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/${file}.c ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/${file}.h)
+endforeach()
+
+
set(teshsuite_src ${teshsuite_src} PARENT_SCOPE)
-set(tesh_files ${tesh_files} PARENT_SCOPE)
+set(tesh_files ${tesh_files} ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/app-bittorrent.tesh PARENT_SCOPE)
+set(bin_files ${bin_files} ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/generate.py PARENT_SCOPE)
set(txt_files ${txt_files} ${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/actions-comm.txt
${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/actions-comm_split_p0.txt
${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/actions-comm_split_p1.txt
set(xml_files ${xml_files} ${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/actions-comm_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/actions-comm/actions-comm_split_d.xml
${CMAKE_CURRENT_SOURCE_DIR}/actions-storage/actions-storage_d.xml
- ${CMAKE_CURRENT_SOURCE_DIR}/trace_integration/test-hbp1.0-hbp1.0-hbp1.0.xml
+ ${CMAKE_CURRENT_SOURCE_DIR}/app-bittorrent/app-bittorrent_d.xml
+ ${CMAKE_CURRENT_SOURCE_DIR}/trace_integration/test-hbp1.0-hbp1.0-hbp1.0.xml
${CMAKE_CURRENT_SOURCE_DIR}/trace_integration/test-hbp1.0-hbp3.0-hbp4.0.xml
${CMAKE_CURRENT_SOURCE_DIR}/trace_integration/test-hbp1.5-hbp1.5.xml
${CMAKE_CURRENT_SOURCE_DIR}/trace_integration/test-hbp1-c0s0-c0s1.xml
ADD_TESH_FACTORIES(tesh-msg-${x} "thread;boost;ucontext;raw" --setenv srcdir=${CMAKE_HOME_DIRECTORY}/teshsuite/msg/${x} --cd ${CMAKE_BINARY_DIR}/teshsuite/msg/${x} ${CMAKE_HOME_DIRECTORY}/teshsuite/msg/${x}/${x}.tesh)
endforeach()
-foreach(x actions-comm actions-storage)
+foreach(x actions-comm actions-storage app-bittorrent)
ADD_TESH_FACTORIES(tesh-msg-${x} "thread;ucontext;raw;boost" --setenv bindir=${CMAKE_BINARY_DIR}/teshsuite/msg/${x} --setenv srcdir=${CMAKE_HOME_DIRECTORY}/examples/platforms --cd ${CMAKE_HOME_DIRECTORY}/teshsuite/msg/${x} ${x}.tesh)
endforeach()
# One context factory is enough for these ones
foreach(x cloud-sharing)
ADD_TESH(tesh-msg-${x} --setenv srcdir=${CMAKE_HOME_DIRECTORY}/teshsuite/msg/${x} --cd ${CMAKE_BINARY_DIR}/teshsuite/msg/${x} ${CMAKE_HOME_DIRECTORY}/teshsuite/msg/${x}/${x}.tesh)
endforeach()
+
+ADD_TESH_FACTORIES(tesh-app-bittorrent-parallel "thread;ucontext;raw" --cfg contexts/nthreads:4 ${CONTEXTS_SYNCHRO} --setenv bindir=${CMAKE_BINARY_DIR}/teshsuite/msg/app-bittorrent --setenv srcdir=${CMAKE_HOME_DIRECTORY}/examples/platforms --cd ${CMAKE_HOME_DIRECTORY}/teshsuite/msg/app-bittorrent app-bittorrent.tesh)
! timeout 10
! output sort 19
-$ $SG_TEST_EXENV ${bindir:=.}/bittorrent ${srcdir:=.}/cluster.xml ${srcdir:=.}/../msg/app-bittorrent/app-bittorrent_d.xml "--log=root.fmt:[%12.6r]%e(%i:%P@%h)%e%m%n"
+$ ${bindir:=.}/bittorrent ${srcdir:=.}/cluster.xml app-bittorrent_d.xml "--log=root.fmt:[%12.6r]%e(%i:%P@%h)%e%m%n"
> [ 0.000000] (1:tracker@node-0.acme.org) Tracker launched.
> [ 0.000000] (2:peer@node-1.acme.org) Hi, I'm joining the network with id 2
> [ 0.000000] (3:peer@node-2.acme.org) Hi, I'm joining the network with id 3
--- /dev/null
+<?xml version='1.0'?>
+<!DOCTYPE platform SYSTEM "http://simgrid.gforge.inria.fr/simgrid/simgrid.dtd">
+<platform version="4.1">
+
+ <actor host="node-0.acme.org" function="tracker">
+ <argument value="3000" />
+ </actor>
+
+ <actor host="node-1.acme.org" function="peer">
+ <argument value="00000002"/> <!-- my id -->
+ <argument value="5000" /> <!-- end time -->
+ <argument value="1" /> <!-- indicates if the peer is a seed at the beginning of the simulation -->
+ </actor>
+ <actor host="node-2.acme.org" function="peer">
+ <argument value="00000003"/> <!-- my id -->
+ <argument value="5000" /> <!-- end time -->
+ </actor>
+ <actor host="node-3.acme.org" function="peer">
+ <argument value="00000004"/> <!-- my id -->
+ <argument value="5000" /> <!-- end time -->
+ </actor>
+ <actor host="node-4.acme.org" function="peer">
+ <argument value="00000005"/> <!-- my id -->
+ <argument value="5000" /> <!-- end time -->
+ <argument value="1" /> <!-- indicates if the peer is a seed at the beginning of the simulation -->
+ </actor>
+ <actor host="node-5.acme.org" function="peer">
+ <argument value="00000006"/> <!-- my id -->
+ <argument value="5000" /> <!-- end time -->
+ </actor>
+ <actor host="node-6.acme.org" function="peer">
+ <argument value="00000007"/> <!-- my id -->
+ <argument value="5000" /> <!-- end time -->
+ </actor>
+ <actor host="node-7.acme.org" function="peer">
+ <argument value="00000008"/> <!-- my id -->
+ <argument value="5000" /> <!-- end time -->
+ </actor>
+</platform>
#include <xbt/RngStream.h>
/** Bittorrent example launcher */
-int main(int argc, char *argv[])
+int main(int argc, char* argv[])
{
msg_host_t host;
unsigned i;
MSG_init(&argc, argv);
/* Check the arguments */
- xbt_assert (argc > 2, "Usage: %s platform_file deployment_file", argv[0]);
+ xbt_assert(argc > 2, "Usage: %s platform_file deployment_file", argv[0]);
MSG_create_environment(argv[1]);
xbt_dynar_t host_list = MSG_hosts_as_dynar();
- xbt_dynar_foreach(host_list, i, host) {
+ xbt_dynar_foreach (host_list, i, host) {
char descr[512];
snprintf(descr, sizeof descr, "RngSream<%s>", MSG_host_get_name(host));
RngStream stream = RngStream_CreateStream(descr);
MSG_main();
- xbt_dynar_foreach(host_list, i, host) {
+ xbt_dynar_foreach (host_list, i, host) {
RngStream stream = (RngStream)MSG_host_get_data(host);
RngStream_DeleteStream(&stream);
MSG_host_set_data(host, NULL);
#define MAILBOX_SIZE 40
#define TRACKER_MAILBOX "tracker_mailbox"
/** Max number of pairs sent by the tracker to clients */
-#define MAXIMUM_PAIRS 50
+#define MAXIMUM_PEERS 50
/** Interval of time where the peer should send a request to the tracker */
#define TRACKER_QUERY_INTERVAL 1000
/** Communication size for a task to the tracker */
/** Number of pieces the peer asks for simultaneously */
#define MAX_PIECES 1
-#endif /* BITTORRENT_BITTORRENT_H_ */
+#endif /* BITTORRENT_BITTORRENT_H_ */
{
connection_t connection = xbt_new(s_connection_t, 1);
- connection->id = id;
- connection->mailbox = bprintf("%d", id);
- connection->bitfield = 0;
- connection->current_piece = -1;
- connection->interested = 0;
- connection->am_interested = 0;
- connection->choked_upload = 1;
+ connection->id = id;
+ connection->mailbox = bprintf("%d", id);
+ connection->bitfield = 0;
+ connection->current_piece = -1;
+ connection->interested = 0;
+ connection->am_interested = 0;
+ connection->choked_upload = 1;
connection->choked_download = 1;
- connection->peer_speed = 0;
- connection->last_unchoke = 0;
+ connection->peer_speed = 0;
+ connection->last_unchoke = 0;
return connection;
}
connection->peer_speed = connection->peer_speed * 0.6 + speed * 0.4;
}
-void connection_free(void *data)
+void connection_free(void* data)
{
- connection_t co = (connection_t) data;
+ connection_t co = (connection_t)data;
xbt_free(co->mailbox);
xbt_free(co);
}
-int connection_has_piece(connection_t connection, unsigned int piece){
- return (connection->bitfield & 1U<<piece);
+int connection_has_piece(connection_t connection, unsigned int piece)
+{
+ return (connection->bitfield & 1U << piece);
}
/** Contains the connection data of a peer. */
typedef struct s_connection {
- int id; //Peer id
- unsigned int bitfield; //Fields
- char *mailbox;
+ int id; // Peer id
+ unsigned int bitfield; // Fields
+ char* mailbox;
int messages_count;
double peer_speed;
double last_unchoke;
int current_piece;
- unsigned int am_interested:1; //Indicates if we are interested in something the peer has
- unsigned int interested:1; //Indicates if the peer is interested in one of our pieces
- unsigned int choked_upload:1; //Indicates if the peer is choked for the current peer
- unsigned int choked_download:1; //Indicates if the peer has choked the current peer
+ unsigned int am_interested : 1; // Indicates if we are interested in something the peer has
+ unsigned int interested : 1; // Indicates if the peer is interested in one of our pieces
+ unsigned int choked_upload : 1; // Indicates if the peer is choked for the current peer
+ unsigned int choked_download : 1; // Indicates if the peer has choked the current peer
} s_connection_t;
-typedef s_connection_t *connection_t;
+typedef s_connection_t* connection_t;
/** @brief Build a new connection object from the peer id.
* @param id id of the peer
*/
void connection_add_speed_value(connection_t connection, double speed);
/** Frees a connection object */
-void connection_free(void *data);
+void connection_free(void* data);
int connection_has_piece(connection_t connection, unsigned int piece);
-#endif /* BITTORRENT_CONNECTION_H_ */
+#endif /* BITTORRENT_CONNECTION_H_ */
--- /dev/null
+/* Copyright (c) 2012-2014. The SimGrid Team.
+ * All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#include "messages.h"
+#include "bittorrent.h"
+
+XBT_LOG_NEW_DEFAULT_CATEGORY(msg_messages, "Messages specific for the message factory");
+
+#define BITS_TO_BYTES(x) (((x) / 8 + (x) % 8) ? 1 : 0)
+
+/** @brief Build a new empty message
+ * @param type type of the message
+ * @param issuer_host_name hostname of the issuer, for debugging purposes
+ * @param mailbox mailbox where the peer should answer
+ * @param peer_id id of the issuer
+ * @param size message size in bytes
+ */
+msg_task_t task_message_new(e_message_type type, const char* issuer_host_name, const char* mailbox, int peer_id,
+ int size)
+{
+ message_t message = xbt_new(s_message_t, 1);
+ message->issuer_host_name = issuer_host_name;
+ message->peer_id = peer_id;
+ message->mailbox = mailbox;
+ message->type = type;
+ msg_task_t task = MSG_task_create(NULL, 0, size, message);
+ XBT_DEBUG("type: %d size: %d", (int)type, size);
+ return task;
+}
+
+/** Builds a message containing an index. */
+msg_task_t task_message_index_new(e_message_type type, const char* issuer_host_name, const char* mailbox, int peer_id,
+ int index, int varsize)
+{
+ msg_task_t task = task_message_new(type, issuer_host_name, mailbox, peer_id, task_message_size(type) + varsize);
+ message_t message = MSG_task_get_data(task);
+ message->index = index;
+ return task;
+}
+
+msg_task_t task_message_bitfield_new(const char* issuer_host_name, const char* mailbox, int peer_id,
+ unsigned int bitfield, int bitfield_size)
+{
+ msg_task_t task = task_message_new(MESSAGE_BITFIELD, issuer_host_name, mailbox, peer_id,
+ task_message_size(MESSAGE_BITFIELD) + BITS_TO_BYTES(bitfield_size));
+ message_t message = MSG_task_get_data(task);
+ message->bitfield = bitfield;
+ return task;
+}
+
+msg_task_t task_message_request_new(const char* issuer_host_name, const char* mailbox, int peer_id, int index,
+ int block_index, int block_length)
+{
+ msg_task_t task = task_message_index_new(MESSAGE_REQUEST, issuer_host_name, mailbox, peer_id, index, 0);
+ message_t message = MSG_task_get_data(task);
+ message->block_index = block_index;
+ message->block_length = block_length;
+ return task;
+}
+
+msg_task_t task_message_piece_new(const char* issuer_host_name, const char* mailbox, int peer_id, int index,
+ int block_index, int block_length, int block_size)
+{
+ msg_task_t task =
+ task_message_index_new(MESSAGE_PIECE, issuer_host_name, mailbox, peer_id, index, block_length * block_size);
+ message_t message = MSG_task_get_data(task);
+ message->block_index = block_index;
+ message->block_length = block_length;
+ return task;
+}
+
+void task_message_free(void* task)
+{
+ message_t message = MSG_task_get_data(task);
+ xbt_free(message);
+ MSG_task_destroy(task);
+}
+
+int task_message_size(e_message_type type)
+{
+ int size = 0;
+ switch (type) {
+ case MESSAGE_HANDSHAKE:
+ size = MESSAGE_HANDSHAKE_SIZE;
+ break;
+ case MESSAGE_CHOKE:
+ size = MESSAGE_CHOKE_SIZE;
+ break;
+ case MESSAGE_UNCHOKE:
+ size = MESSAGE_UNCHOKE_SIZE;
+ break;
+ case MESSAGE_INTERESTED:
+ case MESSAGE_NOTINTERESTED:
+ size = MESSAGE_INTERESTED_SIZE;
+ break;
+ case MESSAGE_HAVE:
+ size = MESSAGE_HAVE_SIZE;
+ break;
+ case MESSAGE_BITFIELD:
+ size = MESSAGE_BITFIELD_SIZE;
+ break;
+ case MESSAGE_REQUEST:
+ size = MESSAGE_REQUEST_SIZE;
+ break;
+ case MESSAGE_PIECE:
+ size = MESSAGE_PIECE_SIZE;
+ break;
+ case MESSAGE_CANCEL:
+ size = MESSAGE_CANCEL_SIZE;
+ break;
+ default:
+ THROW_IMPOSSIBLE;
+ }
+ return size;
+}
/** Message data */
typedef struct s_message {
e_message_type type;
- const char *mailbox;
- const char *issuer_host_name;
+ const char* mailbox;
+ const char* issuer_host_name;
int peer_id;
unsigned int bitfield;
int index;
int block_index;
int block_length;
} s_message_t;
-typedef s_message_t *message_t;
+typedef s_message_t* message_t;
/** Builds a new value-less message */
-msg_task_t task_message_new(e_message_type type, const char *issuer_host_name, const char *mailbox, int peer_id,
+msg_task_t task_message_new(e_message_type type, const char* issuer_host_name, const char* mailbox, int peer_id,
int size);
/** Builds a new "have/piece" message */
-msg_task_t task_message_index_new(e_message_type type, const char *issuer_host_name, const char *mailbox, int peer_id,
+msg_task_t task_message_index_new(e_message_type type, const char* issuer_host_name, const char* mailbox, int peer_id,
int index, int varsize);
/** Builds a new bitfield message */
-msg_task_t task_message_bitfield_new(const char *issuer_host_name, const char *mailbox, int peer_id,
+msg_task_t task_message_bitfield_new(const char* issuer_host_name, const char* mailbox, int peer_id,
unsigned int bitfield, int bitfield_size);
/** Builds a new "request" message */
-msg_task_t task_message_request_new(const char *issuer_host_name, const char *mailbox, int peer_id, int index,
+msg_task_t task_message_request_new(const char* issuer_host_name, const char* mailbox, int peer_id, int index,
int block_index, int block_length);
/** Build a new "piece" message */
-msg_task_t task_message_piece_new(const char *issuer_host_name, const char *mailbox, int peer_id, int index,
+msg_task_t task_message_piece_new(const char* issuer_host_name, const char* mailbox, int peer_id, int index,
int block_index, int block_length, int block_size);
/** Free a message task */
-void task_message_free(void *);
+void task_message_free(void*);
int task_message_size(e_message_type type);
-#endif /* BITTORRENT_MESSAGES_H_ */
+#endif /* BITTORRENT_MESSAGES_H_ */
* under the terms of the license (GNU LGPL) which comes with this package. */
#include "peer.h"
-#include "tracker.h"
#include "connection.h"
#include "messages.h"
+#include "tracker.h"
+#include <limits.h>
#include <simgrid/msg.h>
#include <xbt/RngStream.h>
-#include <limits.h>
XBT_LOG_NEW_DEFAULT_CATEGORY(msg_peers, "Messages specific for the peers");
* User parameters for transferred file data. For the test, the default values are :
* File size: 10 pieces * 5 blocks/piece * 16384 bytes/block = 819200 bytes
*/
-#define FILE_PIECES 10U
-#define PIECES_BLOCKS 5U
-#define BLOCK_SIZE 16384
+#define FILE_PIECES 10UL
+#define PIECES_BLOCKS 5UL
+#define BLOCK_SIZE 16384
static const unsigned long int FILE_SIZE = FILE_PIECES * PIECES_BLOCKS * BLOCK_SIZE;
/** Number of blocks asked by each request */
#define ENABLE_END_GAME_MODE 1
#define SLEEP_DURATION 1
-int count_pieces(unsigned int bitfield){
- int count=0;
+int count_pieces(unsigned int bitfield)
+{
+ int count = 0;
unsigned int n = bitfield;
- while (n){
+ while (n) {
count += n & 1U;
- n >>= 1U ;
+ n >>= 1U;
}
return count;
}
-int peer_has_not_piece(peer_t peer, unsigned int piece){
- return !(peer->bitfield & 1U<<piece);
+int peer_has_not_piece(peer_t peer, unsigned int piece)
+{
+ return !(peer->bitfield & 1U << piece);
}
/** Check that a piece is not currently being download by the peer. */
-int peer_is_not_downloading_piece(peer_t peer, unsigned int piece){
- return !(peer->current_pieces & 1U<<piece);
+int peer_is_not_downloading_piece(peer_t peer, unsigned int piece)
+{
+ return !(peer->current_pieces & 1U << piece);
}
-void get_status(char **status, unsigned int bitfield){
- for(int i=FILE_PIECES-1; i>=0; i--)
+void get_status(char** status, unsigned int bitfield)
+{
+ for (int i = FILE_PIECES - 1; i >= 0; i--)
(*status)[i] = (bitfield & (1U << i)) ? '1' : '0';
(*status)[FILE_PIECES] = '\0';
}
/** Peer main function */
-int peer(int argc, char *argv[])
+int peer(int argc, char* argv[])
{
- //Check arguments
+ // Check arguments
xbt_assert(argc == 3 || argc == 4, "Wrong number of arguments");
- //Build peer object
- peer_t peer = peer_init(xbt_str_parse_int(argv[1],"Invalid ID: %s"), argc==4 ? 1:0);
+ // Build peer object
+ peer_t peer = peer_init(xbt_str_parse_int(argv[1], "Invalid ID: %s"), argc == 4 ? 1 : 0);
- //Retrieve deadline
- double deadline = xbt_str_parse_double(argv[2],"Invalid deadline: %s");
+ // Retrieve deadline
+ double deadline = xbt_str_parse_double(argv[2], "Invalid deadline: %s");
xbt_assert(deadline > 0, "Wrong deadline supplied");
- char *status = xbt_malloc0(FILE_PIECES+1);
+ char* status = xbt_malloc0(FILE_PIECES + 1);
get_status(&status, peer->bitfield);
XBT_INFO("Hi, I'm joining the network with id %d", peer->id);
- //Getting peer data from the tracker.
+ // Getting peer data from the tracker.
if (get_peers_data(peer)) {
XBT_DEBUG("Got %d peers from the tracker. Current status is: %s", xbt_dict_length(peer->peers), status);
peer->begin_receive_time = MSG_get_clock();
handle_message(peer, peer->task_received);
}
} else {
- //We don't execute the choke algorithm if we don't already have a piece
+ // We don't execute the choke algorithm if we don't already have a piece
if (MSG_get_clock() >= next_choked_update && count_pieces(peer->bitfield) > 0) {
update_choked_peers(peer);
next_choked_update += UPDATE_CHOKED_INTERVAL;
{
double next_choked_update = MSG_get_clock() + UPDATE_CHOKED_INTERVAL;
XBT_DEBUG("Start seeding.");
- //start the main seed loop
+ // start the main seed loop
while (MSG_get_clock() < deadline) {
if (peer->comm_received == NULL) {
peer->task_received = NULL;
} else {
if (MSG_get_clock() >= next_choked_update) {
update_choked_peers(peer);
- //TODO: Change the choked peer algorithm when seeding.
+ // TODO: Change the choked peer algorithm when seeding.
next_choked_update += UPDATE_CHOKED_INTERVAL;
} else {
MSG_process_sleep(SLEEP_DURATION);
*/
int get_peers_data(peer_t peer)
{
- int success = 0;
+ int success = 0;
double timeout = MSG_get_clock() + GET_PEERS_TIMEOUT;
- //Build the task to send to the tracker
- tracker_task_data_t data = tracker_task_data_new(MSG_host_get_name(MSG_host_self()), peer->mailbox_tracker,
- peer->id, 0, 0, FILE_SIZE);
+ // Build the task to send to the tracker
+ tracker_task_data_t data =
+ tracker_task_data_new(MSG_host_get_name(MSG_host_self()), peer->mailbox_tracker, peer->id, 0, 0, FILE_SIZE);
msg_task_t task_send = MSG_task_create(NULL, 0, TRACKER_COMM_SIZE, data);
while ((success == 0) && MSG_get_clock() < timeout) {
XBT_DEBUG("Sending a peer request to the tracker.");
}
}
- success = 0;
+ success = 0;
msg_task_t task_received = NULL;
while ((success == 0) && MSG_get_clock() < timeout) {
msg_comm_t comm_received = MSG_task_irecv(&task_received, peer->mailbox_tracker);
- msg_error_t status = MSG_comm_wait(comm_received, GET_PEERS_TIMEOUT);
+ msg_error_t status = MSG_comm_wait(comm_received, GET_PEERS_TIMEOUT);
if (status == MSG_OK) {
tracker_task_data_t data = MSG_task_get_data(task_received);
unsigned i;
int peer_id;
- //Add the peers the tracker gave us to our peer list.
- xbt_dynar_foreach(data->peers, i, peer_id) {
+ // Add the peers the tracker gave us to our peer list.
+ xbt_dynar_foreach (data->peers, i, peer_id) {
if (peer_id != peer->id)
- xbt_dict_set_ext(peer->peers, (char *) &peer_id, sizeof(int), connection_new(peer_id), NULL);
+ xbt_dict_set_ext(peer->peers, (char*)&peer_id, sizeof(int), connection_new(peer_id), NULL);
}
success = 1;
- //free the communication and the task
+ // free the communication and the task
MSG_comm_destroy(comm_received);
tracker_task_data_free(data);
MSG_task_destroy(task_received);
*/
peer_t peer_init(int id, int seed)
{
- peer_t peer = xbt_new(s_peer_t,1);
- peer->id = id;
+ peer_t peer = xbt_new(s_peer_t, 1);
+ peer->id = id;
peer->hostname = MSG_host_get_name(MSG_host_self());
- snprintf(peer->mailbox,MAILBOX_SIZE-1, "%d", id);
- snprintf(peer->mailbox_tracker,MAILBOX_SIZE-1, "tracker_%d", id);
+ snprintf(peer->mailbox, MAILBOX_SIZE - 1, "%d", id);
+ snprintf(peer->mailbox_tracker, MAILBOX_SIZE - 1, "tracker_%d", id);
peer->peers = xbt_dict_new_homogeneous(NULL);
peer->active_peers = xbt_dict_new_homogeneous(NULL);
if (seed) {
- peer->bitfield = (1U<<FILE_PIECES)-1U;
- peer->bitfield_blocks = (1ULL<<(FILE_PIECES * PIECES_BLOCKS))-1ULL ;
+ peer->bitfield = (1U << FILE_PIECES) - 1U;
+ peer->bitfield_blocks = (1ULL << (FILE_PIECES * PIECES_BLOCKS)) - 1ULL;
} else {
- peer->bitfield = 0;
+ peer->bitfield = 0;
peer->bitfield_blocks = 0;
}
/** Destroys a poor peer object. */
void peer_free(peer_t peer)
{
- char *key;
+ char* key;
connection_t connection;
xbt_dict_cursor_t cursor;
- xbt_dict_foreach(peer->peers, cursor, key, connection) {
+ xbt_dict_foreach (peer->peers, cursor, key, connection) {
connection_free(connection);
}
xbt_dict_free(&peer->peers);
*/
int has_finished(unsigned int bitfield)
{
- return bitfield == (1U<<FILE_PIECES)-1U;
+ return bitfield == (1U << FILE_PIECES) - 1U;
}
int nb_interested_peers(peer_t peer)
{
xbt_dict_cursor_t cursor = NULL;
- char *key;
+ char* key;
connection_t connection;
int nb = 0;
- xbt_dict_foreach(peer->peers, cursor, key, connection) {
+ xbt_dict_foreach (peer->peers, cursor, key, connection) {
if (connection->interested)
nb++;
}
void update_active_peers_set(peer_t peer, connection_t remote_peer)
{
if ((remote_peer->interested != 0) && (remote_peer->choked_upload == 0)) {
- //add in the active peers set
- xbt_dict_set_ext(peer->active_peers, (char *) &remote_peer->id, sizeof(int), remote_peer, NULL);
- } else if (xbt_dict_get_or_null_ext(peer->active_peers, (char *) &remote_peer->id, sizeof(int))) {
- xbt_dict_remove_ext(peer->active_peers, (char *) &remote_peer->id, sizeof(int));
+ // add in the active peers set
+ xbt_dict_set_ext(peer->active_peers, (char*)&remote_peer->id, sizeof(int), remote_peer, NULL);
+ } else if (xbt_dict_get_or_null_ext(peer->active_peers, (char*)&remote_peer->id, sizeof(int))) {
+ xbt_dict_remove_ext(peer->active_peers, (char*)&remote_peer->id, sizeof(int));
}
}
*/
void handle_message(peer_t peer, msg_task_t task)
{
- const char* type_names[10] =
- {"HANDSHAKE","CHOKE","UNCHOKE","INTERESTED","NOTINTERESTED","HAVE","BITFIELD","REQUEST", "PIECE", "CANCEL" };
+ const char* type_names[10] = {"HANDSHAKE", "CHOKE", "UNCHOKE", "INTERESTED", "NOTINTERESTED",
+ "HAVE", "BITFIELD", "REQUEST", "PIECE", "CANCEL"};
message_t message = MSG_task_get_data(task);
XBT_DEBUG("Received a %s message from %s (%s)", type_names[message->type], message->mailbox,
- message->issuer_host_name);
+ message->issuer_host_name);
connection_t remote_peer;
- remote_peer = xbt_dict_get_or_null_ext(peer->peers, (char *) &message->peer_id, sizeof(int));
+ remote_peer = xbt_dict_get_or_null_ext(peer->peers, (char*)&message->peer_id, sizeof(int));
switch (message->type) {
- case MESSAGE_HANDSHAKE:
- //Check if the peer is in our connection list.
- if (remote_peer == 0) {
- xbt_dict_set_ext(peer->peers, (char *) &message->peer_id, sizeof(int), connection_new(message->peer_id), NULL);
- send_handshake(peer, message->mailbox);
- }
- //Send our bitfield to the peer
- send_bitfield(peer, message->mailbox);
- break;
- case MESSAGE_BITFIELD:
- //Update the pieces list
- update_pieces_count_from_bitfield(peer, message->bitfield);
- //Store the bitfield
- remote_peer->bitfield = message->bitfield;
- xbt_assert(!remote_peer->am_interested, "Should not be interested at first");
- if (is_interested(peer, remote_peer)) {
- remote_peer->am_interested = 1;
- send_interested(peer, message->mailbox);
- }
- break;
- case MESSAGE_INTERESTED:
- xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?");
- //Update the interested state of the peer.
- remote_peer->interested = 1;
- update_active_peers_set(peer, remote_peer);
- break;
- case MESSAGE_NOTINTERESTED:
- xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?");
- remote_peer->interested = 0;
- update_active_peers_set(peer, remote_peer);
- break;
- case MESSAGE_UNCHOKE:
- xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?");
- xbt_assert(remote_peer->choked_download);
- remote_peer->choked_download = 0;
- //Send requests to the peer, since it has unchoked us
- if (remote_peer->am_interested)
- request_new_piece_to_peer(peer, remote_peer);
- break;
- case MESSAGE_CHOKE:
- xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?");
- xbt_assert(!remote_peer->choked_download);
- remote_peer->choked_download = 1;
- if (remote_peer->current_piece != -1)
- remove_current_piece(peer, remote_peer, remote_peer->current_piece);
- break;
- case MESSAGE_HAVE:
- XBT_DEBUG("\t for piece %d", message->index);
- xbt_assert((message->index >= 0 && message->index < FILE_PIECES), "Wrong HAVE message received");
- remote_peer->bitfield = remote_peer->bitfield | (1U<<message->index);
- peer->pieces_count[message->index]++;
- //If the piece is in our pieces, we tell the peer that we are interested.
- if ((remote_peer->am_interested == 0) && peer_has_not_piece(peer,message->index)) {
- remote_peer->am_interested = 1;
- send_interested(peer, message->mailbox);
- if (remote_peer->choked_download == 0)
+ case MESSAGE_HANDSHAKE:
+ // Check if the peer is in our connection list.
+ if (remote_peer == 0) {
+ xbt_dict_set_ext(peer->peers, (char*)&message->peer_id, sizeof(int), connection_new(message->peer_id), NULL);
+ send_handshake(peer, message->mailbox);
+ }
+ // Send our bitfield to the peer
+ send_bitfield(peer, message->mailbox);
+ break;
+ case MESSAGE_BITFIELD:
+ // Update the pieces list
+ update_pieces_count_from_bitfield(peer, message->bitfield);
+ // Store the bitfield
+ remote_peer->bitfield = message->bitfield;
+ xbt_assert(!remote_peer->am_interested, "Should not be interested at first");
+ if (is_interested(peer, remote_peer)) {
+ remote_peer->am_interested = 1;
+ send_interested(peer, message->mailbox);
+ }
+ break;
+ case MESSAGE_INTERESTED:
+ xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?");
+ // Update the interested state of the peer.
+ remote_peer->interested = 1;
+ update_active_peers_set(peer, remote_peer);
+ break;
+ case MESSAGE_NOTINTERESTED:
+ xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?");
+ remote_peer->interested = 0;
+ update_active_peers_set(peer, remote_peer);
+ break;
+ case MESSAGE_UNCHOKE:
+ xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?");
+ xbt_assert(remote_peer->choked_download);
+ remote_peer->choked_download = 0;
+ // Send requests to the peer, since it has unchoked us
+ if (remote_peer->am_interested)
request_new_piece_to_peer(peer, remote_peer);
- }
- break;
- case MESSAGE_REQUEST:
- xbt_assert(remote_peer->interested);
- xbt_assert((message->index >= 0 && message->index < FILE_PIECES), "Wrong request received");
- if (remote_peer->choked_upload == 0) {
- XBT_DEBUG("\t for piece %d (%d,%d)", message->index, message->block_index,
- message->block_index + message->block_length);
- if (!peer_has_not_piece(peer, message->index)) {
- send_piece(peer, message->mailbox, message->index, message->block_index, message->block_length);
+ break;
+ case MESSAGE_CHOKE:
+ xbt_assert((remote_peer != NULL), "A non-in-our-list peer has sent us a message. WTH ?");
+ xbt_assert(!remote_peer->choked_download);
+ remote_peer->choked_download = 1;
+ if (remote_peer->current_piece != -1)
+ remove_current_piece(peer, remote_peer, remote_peer->current_piece);
+ break;
+ case MESSAGE_HAVE:
+ XBT_DEBUG("\t for piece %d", message->index);
+ xbt_assert((message->index >= 0 && message->index < FILE_PIECES), "Wrong HAVE message received");
+ remote_peer->bitfield = remote_peer->bitfield | (1U << message->index);
+ peer->pieces_count[message->index]++;
+ // If the piece is in our pieces, we tell the peer that we are interested.
+ if ((remote_peer->am_interested == 0) && peer_has_not_piece(peer, message->index)) {
+ remote_peer->am_interested = 1;
+ send_interested(peer, message->mailbox);
+ if (remote_peer->choked_download == 0)
+ request_new_piece_to_peer(peer, remote_peer);
}
- } else {
- XBT_DEBUG("\t for piece %d but he is choked.", message->peer_id);
- }
- break;
- case MESSAGE_PIECE:
- XBT_DEBUG(" \t for piece %d (%d,%d)", message->index, message->block_index,
- message->block_index + message->block_length);
- xbt_assert(!remote_peer->choked_download);
- xbt_assert(remote_peer->am_interested || ENABLE_END_GAME_MODE,
- "Can't received a piece if I'm not interested wihtout end-game mode!"
- "piece (%d) bitfield(%u) remote bitfield(%u)", message->index, peer->bitfield, remote_peer->bitfield);
- xbt_assert(remote_peer->choked_download != 1, "Can't received a piece if I'm choked !");
- xbt_assert((message->index >= 0 && message->index < FILE_PIECES), "Wrong piece received");
- //TODO: Execute à computation.
- if (peer_has_not_piece(peer,message->index)) {
- update_bitfield_blocks(peer, message->index, message->block_index, message->block_length);
- if (piece_complete(peer, message->index)) {
- //Removing the piece from our piece list
- remove_current_piece(peer, remote_peer, message->index);
- //Setting the fact that we have the piece
- peer->bitfield = peer->bitfield | (1U<<message->index);
- char* status = xbt_malloc0(FILE_PIECES+1);
- get_status(&status, peer->bitfield);
- XBT_DEBUG("My status is now %s", status);
- xbt_free(status);
- //Sending the information to all the peers we are connected to
- send_have(peer, message->index);
- //sending UNINTERESTED to peers that do not have what we want.
- update_interested_after_receive(peer);
- } else { // piece not completed
- send_request_to_peer(peer, remote_peer, message->index); // ask for the next block
+ break;
+ case MESSAGE_REQUEST:
+ xbt_assert(remote_peer->interested);
+ xbt_assert((message->index >= 0 && message->index < FILE_PIECES), "Wrong request received");
+ if (remote_peer->choked_upload == 0) {
+ XBT_DEBUG("\t for piece %d (%d,%d)", message->index, message->block_index,
+ message->block_index + message->block_length);
+ if (!peer_has_not_piece(peer, message->index)) {
+ send_piece(peer, message->mailbox, message->index, message->block_index, message->block_length);
+ }
+ } else {
+ XBT_DEBUG("\t for piece %d but he is choked.", message->peer_id);
}
- } else {
- XBT_DEBUG("However, we already have it");
- xbt_assert(ENABLE_END_GAME_MODE, "Should not happen because we don't use end game mode !");
- request_new_piece_to_peer(peer, remote_peer);
- }
- break;
- case MESSAGE_CANCEL:
- break;
- default:
- THROW_IMPOSSIBLE;
+ break;
+ case MESSAGE_PIECE:
+ XBT_DEBUG(" \t for piece %d (%d,%d)", message->index, message->block_index,
+ message->block_index + message->block_length);
+ xbt_assert(!remote_peer->choked_download);
+ xbt_assert(remote_peer->am_interested || ENABLE_END_GAME_MODE,
+ "Can't received a piece if I'm not interested wihtout end-game mode!"
+ "piece (%d) bitfield(%u) remote bitfield(%u)",
+ message->index, peer->bitfield, remote_peer->bitfield);
+ xbt_assert(remote_peer->choked_download != 1, "Can't received a piece if I'm choked !");
+ xbt_assert((message->index >= 0 && message->index < FILE_PIECES), "Wrong piece received");
+ // TODO: Execute à computation.
+ if (peer_has_not_piece(peer, message->index)) {
+ update_bitfield_blocks(peer, message->index, message->block_index, message->block_length);
+ if (piece_complete(peer, message->index)) {
+ // Removing the piece from our piece list
+ remove_current_piece(peer, remote_peer, message->index);
+ // Setting the fact that we have the piece
+ peer->bitfield = peer->bitfield | (1U << message->index);
+ char* status = xbt_malloc0(FILE_PIECES + 1);
+ get_status(&status, peer->bitfield);
+ XBT_DEBUG("My status is now %s", status);
+ xbt_free(status);
+ // Sending the information to all the peers we are connected to
+ send_have(peer, message->index);
+ // sending UNINTERESTED to peers that do not have what we want.
+ update_interested_after_receive(peer);
+ } else { // piece not completed
+ send_request_to_peer(peer, remote_peer, message->index); // ask for the next block
+ }
+ } else {
+ XBT_DEBUG("However, we already have it");
+ xbt_assert(ENABLE_END_GAME_MODE, "Should not happen because we don't use end game mode !");
+ request_new_piece_to_peer(peer, remote_peer);
+ }
+ break;
+ case MESSAGE_CANCEL:
+ break;
+ default:
+ THROW_IMPOSSIBLE;
}
- //Update the peer speed.
+ // Update the peer speed.
if (remote_peer) {
connection_add_speed_value(remote_peer, 1.0 / (MSG_get_clock() - peer->begin_receive_time));
}
{
int piece = select_piece_to_download(peer, remote_peer);
if (piece != -1) {
- peer->current_pieces|= (1U << (unsigned int) piece);
+ peer->current_pieces |= (1U << (unsigned int)piece);
send_request_to_peer(peer, remote_peer, piece);
}
}
if (count_pieces(peer->current_pieces) >= (FILE_PIECES - count_pieces(peer->bitfield)) &&
(is_interested(peer, remote_peer) != 0)) {
#if ENABLE_END_GAME_MODE == 0
- return -1;
+ return -1;
#endif
int nb_interesting_pieces = 0;
// compute the number of interesting pieces
for (int i = 0; i < FILE_PIECES; i++) {
- if (peer_has_not_piece(peer,i) && connection_has_piece(remote_peer,i)) {
+ if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer, i)) {
nb_interesting_pieces++;
}
}
xbt_assert(nb_interesting_pieces != 0);
// get a random interesting piece
int random_piece_index = RngStream_RandInt(peer->stream, 0, nb_interesting_pieces - 1);
- int current_index = 0;
+ int current_index = 0;
for (int i = 0; i < FILE_PIECES; i++) {
- if (peer_has_not_piece(peer,i) && connection_has_piece(remote_peer,i)) {
+ if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer, i)) {
if (random_piece_index == current_index) {
piece = i;
break;
int nb_interesting_pieces = 0;
// compute the number of interesting pieces
for (int i = 0; i < FILE_PIECES; i++) {
- if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) && peer_is_not_downloading_piece(peer, i)) {
+ if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer, i) &&
+ peer_is_not_downloading_piece(peer, i)) {
nb_interesting_pieces++;
}
}
xbt_assert(nb_interesting_pieces != 0);
// get a random interesting piece
int random_piece_index = RngStream_RandInt(peer->stream, 0, nb_interesting_pieces - 1);
- int current_index = 0;
+ int current_index = 0;
for (int i = 0; i < FILE_PIECES; i++) {
- if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) && peer_is_not_downloading_piece(peer, i)) {
+ if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer, i) &&
+ peer_is_not_downloading_piece(peer, i)) {
if (random_piece_index == current_index) {
piece = i;
break;
}
xbt_assert(piece != -1);
return piece;
- } else { // Rarest first policy
- short min = SHRT_MAX;
+ } else { // Rarest first policy
+ short min = SHRT_MAX;
int nb_min_pieces = 0;
int current_index = 0;
// compute the smallest number of copies of available pieces
for (int i = 0; i < FILE_PIECES; i++) {
- if (peer->pieces_count[i] < min && peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) &&
+ if (peer->pieces_count[i] < min && peer_has_not_piece(peer, i) && connection_has_piece(remote_peer, i) &&
peer_is_not_downloading_piece(peer, i))
min = peer->pieces_count[i];
}
- xbt_assert(min != SHRT_MAX || (is_interested_and_free(peer, remote_peer) ==0));
+ xbt_assert(min != SHRT_MAX || (is_interested_and_free(peer, remote_peer) == 0));
// compute the number of rarest pieces
for (int i = 0; i < FILE_PIECES; i++) {
- if (peer->pieces_count[i] == min && peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) &&
+ if (peer->pieces_count[i] == min && peer_has_not_piece(peer, i) && connection_has_piece(remote_peer, i) &&
peer_is_not_downloading_piece(peer, i))
nb_min_pieces++;
}
- xbt_assert(nb_min_pieces != 0 || (is_interested_and_free(peer, remote_peer)==0));
+ xbt_assert(nb_min_pieces != 0 || (is_interested_and_free(peer, remote_peer) == 0));
// get a random rarest piece
int random_rarest_index = RngStream_RandInt(peer->stream, 0, nb_min_pieces - 1);
for (int i = 0; i < FILE_PIECES; i++) {
- if (peer->pieces_count[i] == min && peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) &&
+ if (peer->pieces_count[i] == min && peer_has_not_piece(peer, i) && connection_has_piece(remote_peer, i) &&
peer_is_not_downloading_piece(peer, i)) {
if (random_rarest_index == current_index) {
piece = i;
if (nb_interested_peers(peer) == 0)
return;
XBT_DEBUG("(%d) update_choked peers %u active peers", peer->id, xbt_dict_size(peer->active_peers));
- //update the current round
+ // update the current round
peer->round = (peer->round + 1) % 3;
- char *key;
- char *key_choked=NULL;
+ char* key;
+ char* key_choked = NULL;
connection_t peer_choosed = NULL;
- connection_t peer_choked = NULL;
- //remove a peer from the list
+ connection_t peer_choked = NULL;
+ // remove a peer from the list
xbt_dict_cursor_t cursor = NULL;
xbt_dict_cursor_first(peer->active_peers, &cursor);
if (xbt_dict_length(peer->active_peers) > 0) {
- key_choked = xbt_dict_cursor_get_key(cursor);
+ key_choked = xbt_dict_cursor_get_key(cursor);
peer_choked = xbt_dict_cursor_get_data(cursor);
}
xbt_dict_cursor_free(&cursor);
connection_t connection;
double unchoke_time = MSG_get_clock() + 1;
- xbt_dict_foreach(peer->peers, cursor, key, connection) {
- if (connection->last_unchoke < unchoke_time &&
- (connection->interested != 0) && (connection->choked_upload != 0)) {
+ xbt_dict_foreach (peer->peers, cursor, key, connection) {
+ if (connection->last_unchoke < unchoke_time && (connection->interested != 0) &&
+ (connection->choked_upload != 0)) {
unchoke_time = connection->last_unchoke;
peer_choosed = connection;
}
}
} else {
- //Random optimistic unchoking
+ // Random optimistic unchoking
if (peer->round == 0) {
int j = 0;
do {
- //We choose a random peer to unchoke.
+ // We choose a random peer to unchoke.
int id_chosen = RngStream_RandInt(peer->stream, 0, xbt_dict_length(peer->peers) - 1);
- int i = 0;
+ int i = 0;
connection_t connection;
- xbt_dict_foreach(peer->peers, cursor, key, connection) {
+ xbt_dict_foreach (peer->peers, cursor, key, connection) {
if (i == id_chosen) {
peer_choosed = connection;
break;
else
XBT_DEBUG("Nothing to do, keep going");
j++;
- } while (peer_choosed == NULL && j < MAXIMUM_PAIRS);
+ } while (peer_choosed == NULL && j < MAXIMUM_PEERS);
} else {
- //Use the "fastest download" policy.
+ // Use the "fastest download" policy.
connection_t connection;
double fastest_speed = 0.0;
- xbt_dict_foreach(peer->peers, cursor, key, connection) {
- if (connection->peer_speed > fastest_speed &&
- (connection->choked_upload != 0) && (connection->interested != 0)) {
- peer_choosed = connection;
+ xbt_dict_foreach (peer->peers, cursor, key, connection) {
+ if (connection->peer_speed > fastest_speed && (connection->choked_upload != 0) &&
+ (connection->interested != 0)) {
+ peer_choosed = connection;
fastest_speed = connection->peer_speed;
}
}
}
if (peer_choosed != NULL)
- XBT_DEBUG("(%d) update_choked peers unchoked (%d) ; int (%d) ; choked (%d) ",
- peer->id, peer_choosed->id, peer_choosed->interested, peer_choosed->choked_upload);
+ XBT_DEBUG("(%d) update_choked peers unchoked (%d) ; int (%d) ; choked (%d) ", peer->id, peer_choosed->id,
+ peer_choosed->interested, peer_choosed->choked_upload);
if (peer_choked != peer_choosed) {
if (peer_choked != NULL) {
xbt_assert((!peer_choked->choked_upload), "Tries to choked a choked peer");
peer_choked->choked_upload = 1;
- xbt_assert((*((int *) key_choked) == peer_choked->id));
+ xbt_assert((*((int*)key_choked) == peer_choked->id));
update_active_peers_set(peer, peer_choked);
XBT_DEBUG("(%d) Sending a CHOKE to %d", peer->id, peer_choked->id);
send_choked(peer, peer_choked->mailbox);
if (peer_choosed != NULL) {
xbt_assert((peer_choosed->choked_upload), "Tries to unchoked an unchoked peer");
peer_choosed->choked_upload = 0;
- xbt_dict_set_ext(peer->active_peers, (char *) &peer_choosed->id, sizeof(int), peer_choosed, NULL);
+ xbt_dict_set_ext(peer->active_peers, (char*)&peer_choosed->id, sizeof(int), peer_choosed, NULL);
peer_choosed->last_unchoke = MSG_get_clock();
XBT_DEBUG("(%d) Sending a UNCHOKE to %d", peer->id, peer_choosed->id);
update_active_peers_set(peer, peer_choosed);
*/
void update_interested_after_receive(peer_t peer)
{
- char *key;
+ char* key;
xbt_dict_cursor_t cursor;
connection_t connection;
- xbt_dict_foreach(peer->peers, cursor, key, connection) {
+ xbt_dict_foreach (peer->peers, cursor, key, connection) {
if (connection->am_interested != 0) {
int interested = 0;
- //Check if the peer still has a piece we want.
+ // Check if the peer still has a piece we want.
for (int i = 0; i < FILE_PIECES; i++) {
- if (peer_has_not_piece(peer, i) && connection_has_piece(connection,i)) {
+ if (peer_has_not_piece(peer, i) && connection_has_piece(connection, i)) {
interested = 1;
break;
}
}
- if (!interested) { //no more piece to download from connection
+ if (!interested) { // no more piece to download from connection
connection->am_interested = 0;
send_notinterested(peer, connection->mailbox);
}
xbt_assert((index >= 0 && index <= FILE_PIECES), "Wrong piece.");
xbt_assert((block_index >= 0 && block_index <= PIECES_BLOCKS), "Wrong block : %d.", block_index);
for (int i = block_index; i < (block_index + block_length); i++) {
- peer->bitfield_blocks |= (1ULL<<(unsigned int)(index * PIECES_BLOCKS + i));
+ peer->bitfield_blocks |= (1ULL << (unsigned int)(index * PIECES_BLOCKS + i));
}
}
int piece_complete(peer_t peer, int index)
{
for (int i = 0; i < PIECES_BLOCKS; i++) {
- if (!(peer->bitfield_blocks & 1ULL<<(index * PIECES_BLOCKS + i))) {
+ if (!(peer->bitfield_blocks & 1ULL << (index * PIECES_BLOCKS + i))) {
return 0;
}
}
int get_first_block(peer_t peer, int piece)
{
for (int i = 0; i < PIECES_BLOCKS; i++) {
- if (!(peer->bitfield_blocks & 1ULL<<(piece * PIECES_BLOCKS + i))) {
+ if (!(peer->bitfield_blocks & 1ULL << (piece * PIECES_BLOCKS + i))) {
return i;
}
}
/** Indicates if the remote peer has a piece not stored by the local peer */
int is_interested(peer_t peer, connection_t remote_peer)
{
- return remote_peer->bitfield & (peer->bitfield^((1<<FILE_PIECES)-1));
+ return remote_peer->bitfield & (peer->bitfield ^ ((1 << FILE_PIECES) - 1));
}
/** Indicates if the remote peer has a piece not stored by the local peer nor requested by the local peer */
int is_interested_and_free(peer_t peer, connection_t remote_peer)
{
for (int i = 0; i < FILE_PIECES; i++) {
- if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer,i) && peer_is_not_downloading_piece(peer, i)) {
+ if (peer_has_not_piece(peer, i) && connection_has_piece(remote_peer, i) && peer_is_not_downloading_piece(peer, i)) {
return 1;
}
}
void send_request_to_peer(peer_t peer, connection_t remote_peer, int piece)
{
remote_peer->current_piece = piece;
- xbt_assert(connection_has_piece(remote_peer,piece));
+ xbt_assert(connection_has_piece(remote_peer, piece));
int block_index = get_first_block(peer, piece);
if (block_index != -1) {
int block_length = MIN(BLOCKS_REQUESTED, PIECES_BLOCKS - block_index);
* @param peer peer data
* @param mailbox destination mailbox
*/
-void send_interested(peer_t peer, const char *mailbox)
+void send_interested(peer_t peer, const char* mailbox)
{
msg_task_t task = task_message_new(MESSAGE_INTERESTED, peer->hostname, peer->mailbox, peer->id,
task_message_size(MESSAGE_INTERESTED));
* @param peer peer data
* @param mailbox destination mailbox
*/
-void send_notinterested(peer_t peer, const char *mailbox)
+void send_notinterested(peer_t peer, const char* mailbox)
{
msg_task_t task = task_message_new(MESSAGE_NOTINTERESTED, peer->hostname, peer->mailbox, peer->id,
task_message_size(MESSAGE_NOTINTERESTED));
{
connection_t remote_peer;
xbt_dict_cursor_t cursor = NULL;
- char *key;
- xbt_dict_foreach(peer->peers, cursor, key, remote_peer) {
+ char* key;
+ xbt_dict_foreach (peer->peers, cursor, key, remote_peer) {
msg_task_t task = task_message_new(MESSAGE_HANDSHAKE, peer->hostname, peer->mailbox, peer->id,
task_message_size(MESSAGE_HANDSHAKE));
MSG_task_dsend(task, remote_peer->mailbox, task_message_free);
* @param peer peer data
* @param mailbox mailbox where to we send the message
*/
-void send_handshake(peer_t peer, const char *mailbox)
+void send_handshake(peer_t peer, const char* mailbox)
{
XBT_DEBUG("Sending a HANDSHAKE to %s", mailbox);
msg_task_t task = task_message_new(MESSAGE_HANDSHAKE, peer->hostname, peer->mailbox, peer->id,
}
/** Send a "choked" message to a peer. */
-void send_choked(peer_t peer, const char *mailbox)
+void send_choked(peer_t peer, const char* mailbox)
{
XBT_DEBUG("Sending a CHOKE to %s", mailbox);
- msg_task_t task = task_message_new(MESSAGE_CHOKE, peer->hostname, peer->mailbox, peer->id,
- task_message_size(MESSAGE_CHOKE));
+ msg_task_t task =
+ task_message_new(MESSAGE_CHOKE, peer->hostname, peer->mailbox, peer->id, task_message_size(MESSAGE_CHOKE));
MSG_task_dsend(task, mailbox, task_message_free);
}
/** Send a "unchoked" message to a peer */
-void send_unchoked(peer_t peer, const char *mailbox)
+void send_unchoked(peer_t peer, const char* mailbox)
{
XBT_DEBUG("Sending a UNCHOKE to %s", mailbox);
- msg_task_t task = task_message_new(MESSAGE_UNCHOKE, peer->hostname, peer->mailbox, peer->id,
- task_message_size(MESSAGE_UNCHOKE));
+ msg_task_t task =
+ task_message_new(MESSAGE_UNCHOKE, peer->hostname, peer->mailbox, peer->id, task_message_size(MESSAGE_UNCHOKE));
MSG_task_dsend(task, mailbox, task_message_free);
}
XBT_DEBUG("Sending HAVE message to all my peers");
connection_t remote_peer;
xbt_dict_cursor_t cursor = NULL;
- char *key;
- xbt_dict_foreach(peer->peers, cursor, key, remote_peer) {
+ char* key;
+ xbt_dict_foreach (peer->peers, cursor, key, remote_peer) {
msg_task_t task = task_message_index_new(MESSAGE_HAVE, peer->hostname, peer->mailbox, peer->id, piece,
task_message_size(MESSAGE_HAVE));
MSG_task_dsend(task, remote_peer->mailbox, task_message_free);
/** @brief Send a bitfield message to all the peers the peer has.
* @param peer peer data
*/
-void send_bitfield(peer_t peer, const char *mailbox)
+void send_bitfield(peer_t peer, const char* mailbox)
{
XBT_DEBUG("Sending a BITFIELD to %s", mailbox);
msg_task_t task = task_message_bitfield_new(peer->hostname, peer->mailbox, peer->id, peer->bitfield, FILE_PIECES);
}
/** Send a "request" message to a pair, containing a request for a piece */
-void send_request(peer_t peer, const char *mailbox, int piece, int block_index, int block_length)
+void send_request(peer_t peer, const char* mailbox, int piece, int block_index, int block_length)
{
XBT_DEBUG("Sending a REQUEST to %s for piece %d (%d,%d)", mailbox, piece, block_index, block_length);
msg_task_t task = task_message_request_new(peer->hostname, peer->mailbox, peer->id, piece, block_index, block_length);
}
/** Send a "piece" message to a pair, containing a piece of the file */
-void send_piece(peer_t peer, const char *mailbox, int piece, int block_index, int block_length)
+void send_piece(peer_t peer, const char* mailbox, int piece, int block_index, int block_length)
{
XBT_DEBUG("Sending the PIECE %d (%d,%d) to %s", piece, block_index, block_length, mailbox);
xbt_assert(piece >= 0, "Tried to send a piece that doesn't exist.");
- xbt_assert(!peer_has_not_piece(peer,piece), "Tried to send a piece that we doesn't have.");
- msg_task_t task = task_message_piece_new(peer->hostname, peer->mailbox, peer->id, piece, block_index, block_length,
- BLOCK_SIZE);
+ xbt_assert(!peer_has_not_piece(peer, piece), "Tried to send a piece that we doesn't have.");
+ msg_task_t task =
+ task_message_piece_new(peer->hostname, peer->mailbox, peer->id, piece, block_index, block_length, BLOCK_SIZE);
MSG_task_dsend(task, mailbox, task_message_free);
}
#ifndef BITTORRENT_PEER_H
#define BITTORRENT_PEER_H
+#include "bittorrent.h"
+#include "connection.h"
#include <simgrid/msg.h>
+#include <xbt/RngStream.h>
#include <xbt/dict.h>
#include <xbt/dynar.h>
-#include <xbt/RngStream.h>
-#include "connection.h"
-#include "bittorrent.h"
/** Peer data */
typedef struct s_peer {
- int id; //peer id
+ int id; // peer id
- unsigned int bitfield; //list of pieces the peer has.
- unsigned long long bitfield_blocks; //list of blocks the peer has.
- short *pieces_count; //number of peers that have each piece.
+ unsigned int bitfield; // list of pieces the peer has.
+ unsigned long long bitfield_blocks; // list of blocks the peer has.
+ short* pieces_count; // number of peers that have each piece.
- unsigned int current_pieces; //current pieces the peer is downloading
+ unsigned int current_pieces; // current pieces the peer is downloading
- xbt_dict_t peers; //peers list
- xbt_dict_t active_peers; //active peers list
- int round; //current round for the chocking algorithm.
+ xbt_dict_t peers; // peers list
+ xbt_dict_t active_peers; // active peers list
+ int round; // current round for the chocking algorithm.
- char mailbox[MAILBOX_SIZE]; //peer mailbox.
- char mailbox_tracker[MAILBOX_SIZE]; //pair mailbox while communicating with the tracker.
- const char *hostname; //peer hostname
+ char mailbox[MAILBOX_SIZE]; // peer mailbox.
+ char mailbox_tracker[MAILBOX_SIZE]; // pair mailbox while communicating with the tracker.
+ const char* hostname; // peer hostname
- msg_task_t task_received; //current task being received
- msg_comm_t comm_received; //current comm
+ msg_task_t task_received; // current task being received
+ msg_comm_t comm_received; // current comm
- RngStream stream; //RngStream for
+ RngStream stream; // RngStream for
- double begin_receive_time; //time when the receiving communication has begun, useful for calculating host speed.
+ double begin_receive_time; // time when the receiving communication has begun, useful for calculating host speed.
} s_peer_t;
-typedef s_peer_t *peer_t;
+typedef s_peer_t* peer_t;
/** Peer main function */
-int peer(int argc, char *argv[]);
-void get_status(char **status, unsigned int bitfield);
+int peer(int argc, char* argv[]);
+void get_status(char** status, unsigned int bitfield);
int get_peers_data(peer_t peer);
void leech_loop(peer_t peer, double deadline);
void send_handshake_all(peer_t peer);
-void send_interested(peer_t peer, const char *mailbox);
+void send_interested(peer_t peer, const char* mailbox);
-void send_notinterested(peer_t peer, const char *mailbox);
-void send_handshake(peer_t peer, const char *mailbox);
-void send_bitfield(peer_t peer, const char *mailbox);
-void send_choked(peer_t peer, const char *mailbox);
-void send_unchoked(peer_t peer, const char *mailbox);
+void send_notinterested(peer_t peer, const char* mailbox);
+void send_handshake(peer_t peer, const char* mailbox);
+void send_bitfield(peer_t peer, const char* mailbox);
+void send_choked(peer_t peer, const char* mailbox);
+void send_unchoked(peer_t peer, const char* mailbox);
void send_have(peer_t peer, int piece);
-void send_request(peer_t peer, const char *mailbox, int piece, int block_index, int block_length);
-void send_piece(peer_t peer, const char *mailbox, int piece, int block_index, int block_length);
+void send_request(peer_t peer, const char* mailbox, int piece, int block_index, int block_length);
+void send_piece(peer_t peer, const char* mailbox, int piece, int block_index, int block_length);
-#endif /* BITTORRENT_PEER_H */
+#endif /* BITTORRENT_PEER_H */
#include <simgrid/msg.h>
#include <xbt/RngStream.h>
-static void task_free(void *data);
+static void task_free(void* data);
XBT_LOG_NEW_DEFAULT_CATEGORY(msg_tracker, "Messages specific for the tracker");
/**
* @param argc number of arguments
* @param argv arguments
*/
-int tracker(int argc, char *argv[])
+int tracker(int argc, char* argv[])
{
- //Checking arguments
+ // Checking arguments
xbt_assert(argc == 2, "Wrong number of arguments for the tracker.");
- //Retrieving end time
- double deadline = xbt_str_parse_double(argv[1],"Invalid deadline: %s");
+ // Retrieving end time
+ double deadline = xbt_str_parse_double(argv[1], "Invalid deadline: %s");
xbt_assert(deadline > 0, "Wrong deadline supplied");
- RngStream stream = (RngStream) MSG_host_get_data(MSG_host_self());
- //Building peers array
+ RngStream stream = (RngStream)MSG_host_get_data(MSG_host_self());
+ // Building peers array
xbt_dynar_t peers_list = xbt_dynar_new(sizeof(int), NULL);
XBT_INFO("Tracker launched.");
comm_received = MSG_task_irecv(&task_received, TRACKER_MAILBOX);
}
if (MSG_comm_test(comm_received)) {
- //Check for correct status
+ // Check for correct status
if (MSG_comm_get_status(comm_received) == MSG_OK) {
- //Retrieve the data sent by the peer.
+ // Retrieve the data sent by the peer.
tracker_task_data_t data = MSG_task_get_data(task_received);
- //Add the peer to our peer list.
+ // Add the peer to our peer list.
if (is_in_list(peers_list, data->peer_id) == 0) {
xbt_dynar_push_as(peers_list, int, data->peer_id);
}
- //Sending peers to the peer
+ // Sending peers to the peer
int next_peer;
int peers_length = xbt_dynar_length(peers_list);
- for (int i = 0; i < MAXIMUM_PAIRS && i < peers_length; i++) {
+ for (int i = 0; i < MAXIMUM_PEERS && i < peers_length; i++) {
do {
next_peer = xbt_dynar_get_as(peers_list, RngStream_RandInt(stream, 0, peers_length - 1), int);
} while (is_in_list(data->peers, next_peer));
xbt_dynar_push_as(data->peers, int, next_peer);
}
- //setting the interval
+ // setting the interval
data->interval = TRACKER_QUERY_INTERVAL;
- //sending the task back to the peer.
+ // sending the task back to the peer.
MSG_task_dsend(task_received, data->mailbox, task_free);
- //destroy the communication.
+ // destroy the communication.
}
MSG_comm_destroy(comm_received);
comm_received = NULL;
MSG_process_sleep(1);
}
}
- //Free the remaining communication if any
+ // Free the remaining communication if any
if (comm_received) {
MSG_comm_destroy(comm_received);
}
- //Free the peers list
+ // Free the peers list
xbt_dynar_free(&peers_list);
XBT_INFO("Tracker is leaving");
* Build a new task for the tracker.
* @param issuer_host_name Hostname of the issuer. For debugging purposes
*/
-tracker_task_data_t tracker_task_data_new(const char *issuer_host_name, const char *mailbox, int peer_id,
- int uploaded, int downloaded, int left)
+tracker_task_data_t tracker_task_data_new(const char* issuer_host_name, const char* mailbox, int peer_id, int uploaded,
+ int downloaded, int left)
{
tracker_task_data_t task = xbt_new(s_tracker_task_data_t, 1);
- task->type = TRACKER_TASK_QUERY;
+ task->type = TRACKER_TASK_QUERY;
task->issuer_host_name = issuer_host_name;
- task->mailbox = mailbox;
- task->peer_id = peer_id;
- task->uploaded = uploaded;
- task->downloaded = downloaded;
- task->left = left;
+ task->mailbox = mailbox;
+ task->peer_id = peer_id;
+ task->uploaded = uploaded;
+ task->downloaded = downloaded;
+ task->left = left;
task->peers = xbt_dynar_new(sizeof(int), NULL);
* Free a tracker task that has not successfully been sent.
* @param data Task to free
*/
-static void task_free(void *data)
+static void task_free(void* data)
{
tracker_task_data_t task_data = MSG_task_get_data(data);
tracker_task_data_free(task_data);
--- /dev/null
+/* Copyright (c) 2012-2014. The SimGrid Team.
+ * All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#ifndef BITTORRENT_TRACKER_H_
+#define BITTORRENT_TRACKER_H_
+#include "bittorrent.h"
+#include <xbt/dynar.h>
+/**
+ * Tracker main function
+ */
+int tracker(int argc, char* argv[]);
+/**
+ * Task types exchanged between a node and the tracker
+ */
+typedef enum { TRACKER_TASK_QUERY, TRACKER_TASK_ANSWER } e_tracker_task_type_t;
+/**
+ * Tasks exchanged between a tracker and peers.
+ */
+typedef struct s_tracker_task_data {
+ e_tracker_task_type_t type; // type of the task
+ const char* mailbox; // mailbox where the tracker should answer
+ const char* issuer_host_name; // hostname, for debug purposes
+ // Query data
+ int peer_id; // peer id
+ int uploaded; // how much the peer has already uploaded
+ int downloaded; // how much the peer has downloaded
+ int left; // how much the peer has left
+ // Answer data
+ int interval; // how often the peer should contact the tracker (unused for now)
+ xbt_dynar_t peers; // the peer list the peer has asked for.
+} s_tracker_task_data_t;
+typedef s_tracker_task_data_t* tracker_task_data_t;
+
+tracker_task_data_t tracker_task_data_new(const char* issuer_host_name, const char* mailbox, int peer_id, int uploaded,
+ int downloaded, int left);
+void tracker_task_data_free(tracker_task_data_t task);
+
+int is_in_list(xbt_dynar_t peers, int id);
+#endif /* BITTORRENT_TRACKER_H */
static void dump_platform()
{
int version = 4;
- xbt_dict_t props = nullptr;
- xbt_dict_cursor_t cursor = nullptr;
- char* key;
- char* data;
+ std::map<std::string, std::string>* props = nullptr;
std::printf("<?xml version='1.0'?>\n");
std::printf("<!DOCTYPE platform SYSTEM \"http://simgrid.gforge.inria.fr/simgrid/simgrid.dtd\">\n");
for (unsigned int i = 0; i < totalHosts; i++) {
std::printf(" <host id=\"%s\" speed=\"%.0f\"", hosts[i]->getCname(), sg_host_speed(hosts[i]));
- props = sg_host_get_properties(hosts[i]);
+ props = hosts[i]->getProperties();
if (hosts[i]->getCoreCount() > 1) {
std::printf(" core=\"%d\"", hosts[i]->getCoreCount());
}
- if (props && not xbt_dict_is_empty(props)) {
+ if (props && not props->empty()) {
std::printf(">\n");
- xbt_dict_foreach (props, cursor, key, data) {
- std::printf(" <prop id=\"%s\" value=\"%s\"/>\n", key, data);
+ for (auto kv : *props) {
+ std::printf(" <prop id=\"%s\" value=\"%s\"/>\n", kv.first.c_str(), kv.second.c_str());
}
std::printf(" </host>\n");
} else {
std::printf("/>\n");
}
- xbt_dict_free(&props);
}
// Routers
-foreach(x heap_bench log_large log_usage mallocator parallel_log_crashtest parmap_bench parmap_test)
+foreach(x heap_bench log_large log_usage mallocator parallel_log_crashtest)
add_executable (${x} ${x}/${x}.c)
target_link_libraries(${x} simgrid)
set_target_properties(${x} PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/${x})
set(teshsuite_src ${teshsuite_src} ${CMAKE_CURRENT_SOURCE_DIR}/${x}/${x}.c)
endforeach()
+foreach(x parmap_bench parmap_test)
+ add_executable (${x} ${x}/${x}.cpp)
+ target_link_libraries(${x} simgrid)
+ set_target_properties(${x} PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/${x})
+
+ set(tesh_files ${tesh_files} ${CMAKE_CURRENT_SOURCE_DIR}/${x}/${x}.tesh)
+ set(teshsuite_src ${teshsuite_src} ${CMAKE_CURRENT_SOURCE_DIR}/${x}/${x}.cpp)
+endforeach()
+
if(HAVE_MMALLOC)
add_executable (mmalloc_test ${CMAKE_CURRENT_SOURCE_DIR}/mmalloc/mmalloc_test.cpp)
target_link_libraries(mmalloc_test simgrid)
+++ /dev/null
-/* Copyright (c) 2012-2017. The SimGrid Team. All rights reserved. */
-
-/* This program is free software; you can redistribute it and/or modify it
- * under the terms of the license (GNU LGPL) which comes with this package. */
-
-#include "simgrid/msg.h"
-#include "src/internal_config.h" /* HAVE_FUTEX_H */
-#include "xbt/xbt_os_time.h"
-#include <stdio.h>
-#include <stdlib.h>
-#include <xbt/dynar.h>
-#include <xbt/parmap.h>
-#include <xbt/sysdep.h>
-
-#define MODES_DEFAULT 0x7
-#define TIMEOUT 10.0
-#define ARRAY_SIZE 10007
-#define FIBO_MAX 25
-
-void (*fun_to_apply)(void *);
-
-static const char *parmap_mode_name(e_xbt_parmap_mode_t mode)
-{
- static char name[80];
- switch (mode) {
- case XBT_PARMAP_POSIX:
- snprintf(name, sizeof name, "POSIX");
- break;
- case XBT_PARMAP_FUTEX:
- snprintf(name, sizeof name, "FUTEX");
- break;
- case XBT_PARMAP_BUSY_WAIT:
- snprintf(name, sizeof name, "BUSY_WAIT");
- break;
- case XBT_PARMAP_DEFAULT:
- snprintf(name, sizeof name, "DEFAULT");
- break;
- default:
- snprintf(name, sizeof name, "UNKNOWN(%d)", (int)mode);
- break;
- }
- return name;
-}
-
-static int parmap_skip_mode(e_xbt_parmap_mode_t mode)
-{
-#if !HAVE_FUTEX_H
- if (mode == XBT_PARMAP_FUTEX) {
- printf("not available\n");
- return 1;
- } else
-#endif
- return 0;
-}
-
-static unsigned fibonacci(unsigned n)
-{
- if (n < 2)
- return n;
- else
- return fibonacci(n - 1) + fibonacci(n - 2);
-}
-
-static void fun_small_comp(void *arg)
-{
- unsigned *u = arg;
- *u = 2 * *u + 1;
-}
-
-static void fun_big_comp(void *arg)
-{
- unsigned *u = arg;
- *u = fibonacci(*u % FIBO_MAX);
-}
-
-static void array_new(unsigned **a, xbt_dynar_t *data)
-{
- *a = xbt_malloc(ARRAY_SIZE * sizeof **a);
- *data = xbt_dynar_new(sizeof *a, NULL);
- xbt_dynar_shrink(*data, ARRAY_SIZE);
- for (int i = 0 ; i < ARRAY_SIZE ; i++) {
- (*a)[i] = i;
- xbt_dynar_push_as(*data, void*, &(*a)[i]);
- }
-}
-
-static void bench_parmap_full(int nthreads, e_xbt_parmap_mode_t mode)
-{
- unsigned *a;
- xbt_dynar_t data;
- xbt_parmap_t parmap;
- double elapsed_time;
-
- printf("** mode = %-15s ", parmap_mode_name(mode));
- fflush(stdout);
-
- if (parmap_skip_mode(mode))
- return;
-
- array_new(&a, &data);
-
- int i = 0;
- double start_time = xbt_os_time();
- do {
- parmap = xbt_parmap_new(nthreads, mode);
- xbt_parmap_apply(parmap, fun_to_apply, data);
- xbt_parmap_destroy(parmap);
- elapsed_time = xbt_os_time() - start_time;
- i++;
- } while (elapsed_time < TIMEOUT);
-
- printf("ran %d times in %g seconds (%g/s)\n", i, elapsed_time, i / elapsed_time);
-
- xbt_dynar_free(&data);
- xbt_free(a);
-}
-
-static void bench_parmap_apply(int nthreads, e_xbt_parmap_mode_t mode)
-{
- unsigned *a;
- xbt_dynar_t data;
- double elapsed_time;
-
- printf("** mode = %-15s ", parmap_mode_name(mode));
- fflush(stdout);
-
- if (parmap_skip_mode(mode))
- return;
-
- array_new(&a, &data);
-
- xbt_parmap_t parmap = xbt_parmap_new(nthreads, mode);
- int i = 0;
- double start_time = xbt_os_time();
- do {
- xbt_parmap_apply(parmap, fun_to_apply, data);
- elapsed_time = xbt_os_time() - start_time;
- i++;
- } while (elapsed_time < TIMEOUT);
- xbt_parmap_destroy(parmap);
-
- printf("ran %d times in %g seconds (%g/s)\n", i, elapsed_time, i / elapsed_time);
-
- xbt_dynar_free(&data);
- xbt_free(a);
-}
-
-static void bench_all_modes(void (*bench_fun)(int, e_xbt_parmap_mode_t),
- int nthreads, unsigned modes)
-{
- e_xbt_parmap_mode_t all_modes[] = {XBT_PARMAP_POSIX, XBT_PARMAP_FUTEX, XBT_PARMAP_BUSY_WAIT, XBT_PARMAP_DEFAULT};
-
- for (unsigned i = 0 ; i < sizeof all_modes / sizeof all_modes[0] ; i++) {
- if (1U << i & modes)
- bench_fun(nthreads, all_modes[i]);
- }
-}
-
-int main(int argc, char *argv[])
-{
- int nthreads;
- unsigned modes = MODES_DEFAULT;
-
- MSG_init(&argc, argv);
-
- if (argc != 2 && argc != 3) {
- fprintf(stderr, "Usage: %s nthreads [modes]\n"
- " nthreads - number of working threads\n"
- " modes - bitmask of modes to test\n",
- argv[0]);
- return EXIT_FAILURE;
- }
- nthreads = atoi(argv[1]);
- if (nthreads < 1) {
- fprintf(stderr, "ERROR: invalid thread count: %d\n", nthreads);
- return EXIT_FAILURE;
- }
- if (argc == 3)
- modes = strtol(argv[2], NULL, 0);
-
- printf("Parmap benchmark with %d workers (modes = %#x)...\n\n", nthreads, modes);
-
- fun_to_apply = &fun_small_comp;
-
- printf("Benchmark for parmap create+apply+destroy (small comp):\n");
- bench_all_modes(bench_parmap_full, nthreads, modes);
- printf("\n");
-
- printf("Benchmark for parmap apply only (small comp):\n");
- bench_all_modes(bench_parmap_apply, nthreads, modes);
- printf("\n");
-
- fun_to_apply = &fun_big_comp;
-
- printf("Benchmark for parmap create+apply+destroy (big comp):\n");
- bench_all_modes(bench_parmap_full, nthreads, modes);
- printf("\n");
-
- printf("Benchmark for parmap apply only (big comp):\n");
- bench_all_modes(bench_parmap_apply, nthreads, modes);
- printf("\n");
-
- return EXIT_SUCCESS;
-}
--- /dev/null
+/* Copyright (c) 2012-2017. The SimGrid Team. All rights reserved. */
+
+/* This program is free software; you can redistribute it and/or modify it
+ * under the terms of the license (GNU LGPL) which comes with this package. */
+
+#include "src/internal_config.h" // HAVE_FUTEX_H
+#include <simgrid/msg.h>
+#include <xbt.h>
+#include <xbt/parmap.hpp>
+
+#include <cstdlib>
+#include <iomanip>
+#include <iostream>
+#include <numeric> // std::iota
+#include <string>
+#include <vector>
+
+#define MODES_DEFAULT 0x7
+#define TIMEOUT 10.0
+#define ARRAY_SIZE 10007
+#define FIBO_MAX 25
+
+void (*fun_to_apply)(unsigned*);
+
+static std::string parmap_mode_name(e_xbt_parmap_mode_t mode)
+{
+ std::string name;
+ switch (mode) {
+ case XBT_PARMAP_POSIX:
+ name = "POSIX";
+ break;
+ case XBT_PARMAP_FUTEX:
+ name = "FUTEX";
+ break;
+ case XBT_PARMAP_BUSY_WAIT:
+ name = "BUSY_WAIT";
+ break;
+ case XBT_PARMAP_DEFAULT:
+ name = "DEFAULT";
+ break;
+ default:
+ name = "UNKNOWN(" + std::to_string(mode) + ")";
+ break;
+ }
+ return name;
+}
+
+static bool parmap_skip_mode(e_xbt_parmap_mode_t mode)
+{
+ if (mode == XBT_PARMAP_FUTEX && not HAVE_FUTEX_H) {
+ std::cout << "not available\n";
+ return true;
+ } else {
+ return false;
+ }
+}
+
+static unsigned fibonacci(unsigned n)
+{
+ if (n < 2)
+ return n;
+ else
+ return fibonacci(n - 1) + fibonacci(n - 2);
+}
+
+static void fun_small_comp(unsigned* arg)
+{
+ *arg = 2 * *arg + 1;
+}
+
+static void fun_big_comp(unsigned* arg)
+{
+ *arg = fibonacci(*arg % FIBO_MAX);
+}
+
+static void bench_parmap_full(int nthreads, e_xbt_parmap_mode_t mode)
+{
+ std::cout << "** mode = " << std::left << std::setw(15) << parmap_mode_name(mode) << " ";
+ std::cout.flush();
+
+ if (parmap_skip_mode(mode))
+ return;
+
+ std::vector<unsigned> a(ARRAY_SIZE);
+ std::vector<unsigned*> data(ARRAY_SIZE);
+ std::iota(begin(a), end(a), 0);
+ std::iota(begin(data), end(data), &a[0]);
+
+ int i = 0;
+ double start_time = xbt_os_time();
+ double elapsed_time;
+ do {
+ {
+ simgrid::xbt::Parmap<unsigned*> parmap(nthreads, mode);
+ parmap.apply(fun_to_apply, data);
+ } // enclosing block to ensure that the parmap is destroyed here.
+ elapsed_time = xbt_os_time() - start_time;
+ i++;
+ } while (elapsed_time < TIMEOUT);
+
+ std::cout << "ran " << i << " times in " << elapsed_time << " seconds (" << (i / elapsed_time) << "/s)\n";
+}
+
+static void bench_parmap_apply(int nthreads, e_xbt_parmap_mode_t mode)
+{
+ std::cout << "** mode = " << std::left << std::setw(15) << parmap_mode_name(mode) << " ";
+ std::cout.flush();
+
+ if (parmap_skip_mode(mode))
+ return;
+
+ std::vector<unsigned> a(ARRAY_SIZE);
+ std::vector<unsigned*> data(ARRAY_SIZE);
+ std::iota(begin(a), end(a), 0);
+ std::iota(begin(data), end(data), &a[0]);
+
+ simgrid::xbt::Parmap<unsigned*> parmap(nthreads, mode);
+ int i = 0;
+ double start_time = xbt_os_time();
+ double elapsed_time;
+ do {
+ parmap.apply(fun_to_apply, data);
+ elapsed_time = xbt_os_time() - start_time;
+ i++;
+ } while (elapsed_time < TIMEOUT);
+
+ std::cout << "ran " << i << " times in " << elapsed_time << " seconds (" << (i / elapsed_time) << "/s)\n";
+}
+
+static void bench_all_modes(void (*bench_fun)(int, e_xbt_parmap_mode_t), int nthreads, unsigned modes)
+{
+ std::vector<e_xbt_parmap_mode_t> all_modes = {XBT_PARMAP_POSIX, XBT_PARMAP_FUTEX, XBT_PARMAP_BUSY_WAIT,
+ XBT_PARMAP_DEFAULT};
+
+ for (unsigned i = 0; i < all_modes.size(); i++) {
+ if (1U << i & modes)
+ bench_fun(nthreads, all_modes[i]);
+ }
+}
+
+int main(int argc, char* argv[])
+{
+ int nthreads;
+ unsigned modes = MODES_DEFAULT;
+
+ MSG_init(&argc, argv);
+
+ if (argc != 2 && argc != 3) {
+ std::cerr << "Usage: " << argv[0] << " nthreads [modes]\n"
+ << " nthreads - number of working threads\n"
+ << " modes - bitmask of modes to test\n";
+ return EXIT_FAILURE;
+ }
+ nthreads = atoi(argv[1]);
+ if (nthreads < 1) {
+ std::cerr << "ERROR: invalid thread count: " << nthreads << "\n";
+ return EXIT_FAILURE;
+ }
+ if (argc == 3)
+ modes = strtol(argv[2], NULL, 0);
+
+ std::cout << "Parmap benchmark with " << nthreads << " workers (modes = " << std::hex << modes << std::dec
+ << ")...\n\n";
+
+ fun_to_apply = &fun_small_comp;
+
+ std::cout << "Benchmark for parmap create+apply+destroy (small comp):\n";
+ bench_all_modes(bench_parmap_full, nthreads, modes);
+ std::cout << std::endl;
+
+ std::cout << "Benchmark for parmap apply only (small comp):\n";
+ bench_all_modes(bench_parmap_apply, nthreads, modes);
+ std::cout << std::endl;
+
+ fun_to_apply = &fun_big_comp;
+
+ std::cout << "Benchmark for parmap create+apply+destroy (big comp):\n";
+ bench_all_modes(bench_parmap_full, nthreads, modes);
+ std::cout << std::endl;
+
+ std::cout << "Benchmark for parmap apply only (big comp):\n";
+ bench_all_modes(bench_parmap_apply, nthreads, modes);
+ std::cout << std::endl;
+
+ return EXIT_SUCCESS;
+}
/* This program is free software; you can redistribute it and/or modify it
* under the terms of the license (GNU LGPL) which comes with this package. */
-#include "simgrid/msg.h"
-#include "src/internal_config.h"
-#include "xbt.h"
-#include "xbt/ex.h"
-#include "xbt/xbt_os_time.h"
+#include "src/internal_config.h" // HAVE_FUTEX_H
+#include <simgrid/msg.h>
+#include <xbt.h>
+#include <xbt/parmap.hpp>
+
+#include <algorithm>
+#include <cstdlib>
+#include <numeric> // std::iota
+#include <vector>
XBT_LOG_NEW_DEFAULT_CATEGORY(parmap_test, "Test for parmap");
-static void fun_double(void *arg)
+static void fun_double(unsigned* arg)
{
- unsigned *u = arg;
- *u = 2 * *u + 1;
+ *arg = 2 * *arg + 1;
}
static int test_parmap_basic(e_xbt_parmap_mode_t mode)
{
int ret = 0;
- unsigned num_workers;
- for (num_workers = 1 ; num_workers <= 16 ; num_workers *= 2) {
+ for (unsigned num_workers = 1; num_workers <= 16; num_workers *= 2) {
const unsigned len = 1033;
const unsigned num = 5;
- unsigned *a;
- xbt_dynar_t data;
- xbt_parmap_t parmap;
- unsigned i;
-
- parmap = xbt_parmap_new(num_workers, mode);
-
- a = xbt_malloc(len * sizeof *a);
- data = xbt_dynar_new(sizeof a, NULL);
- for (i = 0; i < len; i++) {
- a[i] = i;
- xbt_dynar_push_as(data, void *, &a[i]);
- }
- for (i = 0; i < num; i++)
- xbt_parmap_apply(parmap, fun_double, data);
+ simgrid::xbt::Parmap<unsigned*> parmap(num_workers, mode);
+ std::vector<unsigned> a(len);
+ std::vector<unsigned*> data(len);
+ std::iota(begin(a), end(a), 0);
+ std::iota(begin(data), end(data), &a[0]);
+
+ for (unsigned i = 0; i < num; i++)
+ parmap.apply(fun_double, data);
- for (i = 0; i < len; i++) {
+ for (unsigned i = 0; i < len; i++) {
unsigned expected = (1U << num) * (i + 1) - 1;
if (a[i] != expected) {
XBT_CRITICAL("with %u threads, a[%u]: expected %u, got %u", num_workers, i, expected, a[i]);
break;
}
}
-
- xbt_dynar_free(&data);
- xbt_free(a);
- xbt_parmap_destroy(parmap);
}
return ret;
}
-static void fun_get_id(void *arg)
+static void fun_get_id(uintptr_t* arg)
{
- *(uintptr_t *)arg = (uintptr_t)xbt_os_thread_self();
+ *arg = (uintptr_t)xbt_os_thread_self();
xbt_os_sleep(0.05);
}
-static int fun_compare(const void *pa, const void *pb)
-{
- uintptr_t a = *(uintptr_t *)pa;
- uintptr_t b = *(uintptr_t *)pb;
- return a < b ? -1 : a > b ? 1 : 0;
-}
-
static int test_parmap_extended(e_xbt_parmap_mode_t mode)
{
int ret = 0;
- unsigned num_workers;
- for (num_workers = 1 ; num_workers <= 16 ; num_workers *= 2) {
+ for (unsigned num_workers = 1; num_workers <= 16; num_workers *= 2) {
const unsigned len = 2 * num_workers;
- uintptr_t *a;
- xbt_parmap_t parmap;
- xbt_dynar_t data;
- unsigned i;
- unsigned count;
-
- parmap = xbt_parmap_new(num_workers, mode);
-
- a = xbt_malloc(len * sizeof *a);
- data = xbt_dynar_new(sizeof a, NULL);
- for (i = 0; i < len; i++)
- xbt_dynar_push_as(data, void *, &a[i]);
-
- xbt_parmap_apply(parmap, fun_get_id, data);
-
- qsort(a, len, sizeof a[0], fun_compare);
- count = 1;
- for (i = 1; i < len; i++)
- if (a[i] != a[i - 1])
- count++;
+
+ simgrid::xbt::Parmap<uintptr_t*> parmap(num_workers, mode);
+ std::vector<uintptr_t> a(len);
+ std::vector<uintptr_t*> data(len);
+ std::iota(begin(data), end(data), &a[0]);
+
+ parmap.apply(fun_get_id, data);
+
+ std::sort(begin(a), end(a));
+ unsigned count = std::distance(begin(a), std::unique(begin(a), end(a)));
if (count != num_workers) {
XBT_CRITICAL("only %u/%u threads did some work", count, num_workers);
ret = 1;
}
-
- xbt_dynar_free(&data);
- xbt_free(a);
- xbt_parmap_destroy(parmap);
}
return ret;
}
src/include/instr/instr_interface.h
src/include/mc/datatypes.h
src/include/mc/mc.h
- src/mc/mc_mmu.h
- src/mc/PageStore.hpp
- src/mc/mc_record.h
src/include/simgrid/sg_config.h
src/include/smpi/smpi_utils.hpp
src/include/surf/datatypes.h
src/include/surf/maxmin.h
src/include/surf/surf.h
+ src/include/xbt/parmap.hpp
+ src/mc/mc_mmu.h
+ src/mc/mc_record.h
+ src/mc/PageStore.hpp
src/msg/msg_private.h
src/simdag/dax.dtd
src/simdag/dax_dtd.c
include/smpi/smpi_extended_traces_fortran.h
include/smpi/forward.hpp
include/xbt.h
- include/xbt/RngStream.h
+ include/xbt/algorithm.hpp
include/xbt/asserts.h
include/xbt/automaton.h
include/xbt/automaton.hpp
+ include/xbt/backtrace.h
+ include/xbt/backtrace.hpp
include/xbt/base.h
include/xbt/config.h
include/xbt/config.hpp
include/xbt/cunit.h
include/xbt/dict.h
- include/xbt/string.hpp
- include/xbt/signal.hpp
include/xbt/dynar.h
include/xbt/dynar.hpp
include/xbt/ex.h
include/xbt/ex.hpp
include/xbt/exception.hpp
- include/xbt/backtrace.h
- include/xbt/backtrace.hpp
+ include/xbt/Extendable.hpp
include/xbt/file.h
- include/xbt/function_types.h
include/xbt/functional.hpp
+ include/xbt/function_types.h
include/xbt/future.hpp
include/xbt/graph.h
include/xbt/heap.h
- include/xbt/Extendable.hpp
include/xbt/log.h
include/xbt/log.hpp
include/xbt/mallocator.h
include/xbt/parmap.h
include/xbt/range.hpp
include/xbt/replay.hpp
+ include/xbt/RngStream.h
+ include/xbt/signal.hpp
include/xbt/str.h
+ include/xbt/string.hpp
include/xbt/swag.h
include/xbt/synchro.h
include/xbt/sysdep.h
