+++ /dev/null
-.deps .libs Makefile Makefile.in _*.c alnem_builder alnem_maestro alnem_sensor alnem_simulator
+++ /dev/null
-DISTCLEANFILES=Makefile.in
-INCLUDES= -I$(top_srcdir)/src/include
-#-I../../src -I../../nws_portability/Include @CFLAGS_SimGrid@
-AM_CFLAGS=-g
-
-check_PROGRAMS=alnem_simulator alnem_sensor alnem_maestro
-
-alnem_simulator_SOURCES= _alnem_simulator.c alnem.c
-alnem_sensor_SOURCES= _alnem_sensor.c alnem.c
-alnem_maestro_SOURCES= _alnem_maestro.c alnem.c
-
-
-alnem_simulator_LDADD= $(top_srcdir)/src/base/libgrassg.a $(top_srcdir)/src/modules/libgrasmodules.a @LIBS_SimGrid@
-alnem_sensor_LDADD= $(top_srcdir)/src/base/libgrasrl.a $(top_srcdir)/src/modules/libgrasmodules.a @LIBS_SimGrid@
-alnem_maestro_LDADD= $(top_srcdir)/src/base/libgrasrl.a $(top_srcdir)/src/modules/libgrasmodules.a @LIBS_SimGrid@
-
-bin_PROGRAMS=alnem_builder
-alnem_builder_LDADD= @LIBS_SimGrid@
-
-# Take care of generatated sources
-NAME=alnem
-PROCESSES=sensor maestro
-include $(top_srcdir)/examples/temps-gras-stub.mk
-
-
-
-
+++ /dev/null
-/* $Id$ */
-
-/* ALNeM itself */
-
-/* Copyright (c) 2003 Martin Quinson. 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 <stdio.h>
-#include <stdlib.h>
-#include <signal.h>
-#include <time.h>
-
-#include <gras.h>
-
-#include <tbx_graph.h> /* alvin's graph toolbox (+ reconstruction algorithm) */
-
-/* **********************************************************************
- * Sensor code
- * **********************************************************************/
-
-/* Global private data */
-typedef struct {
- gras_sock_t *sock;
-} sensor_data_t;
-
-/* Function prototypes */
-int sensor (int argc,char *argv[]);
-
-int sensor (int argc,char *argv[]) {
- xbt_error_t errcode;
- sensor_data_t *g=gras_userdata_new(sensor_data_t);
-
- if ((errcode=gras_sock_server_open(4000,4000,&(g->sock)))) {
- fprintf(stderr,"Sensor: Error %s encountered while opening the server socket\n",xbt_error_name(errcode));
- return 1;
- }
-
- if (grasbw_register_messages()) {
- gras_sock_close(g->sock);
- return 1;
- }
-
- while (1) {
- if ((errcode=gras_msg_handle(3600.0)) && errcode != timeout_error) {
- fprintf(stderr,"Sensor: Error '%s' while handling message\n",
- xbt_error_name(errcode));
- }
- }
-
- gras_sleep(5,0);
- return gras_sock_close(g->sock);
-}
-
-/* **********************************************************************
- * Maestro code
- * **********************************************************************/
-
-/* Global private data */
-typedef struct {
- gras_sock_t *sock;
-} maestro_data_t;
-
-/* Function prototypes */
-int maestro (int argc,char *argv[]);
-
-#define MAXHOSTS 100
-
-#define INTERF(graph,table,a,u,b,v) INTERFERENCE(table,\
- TBX_Graph_nodeSearch(graph,a),\
- TBX_Graph_nodeSearch(graph,u),\
- TBX_Graph_nodeSearch(graph,b),\
- TBX_Graph_nodeSearch(graph,v))
-
-int maestro(int argc,char *argv[]) {
- int bufSize=32 * 1024;
- int expSize= 1024 * 1024;
- int msgSize=expSize;
- int satSize=msgSize * 100;
- double dummy,beginSim;
- xbt_error_t errcode;
- maestro_data_t *g=gras_userdata_new(maestro_data_t);
-
- double bw[MAXHOSTS][MAXHOSTS];
- double bw_sat[MAXHOSTS][MAXHOSTS];
-
- int a,b,c,d,begin;
-
- TBX_Graph_t graph = TBX_Graph_newGraph("Essai",0,NULL); /* a dummy graph containing all hosts */
- TBX_FIFO_t host_fifo = TBX_FIFO_newFIFO();
- TBX_InterfTable_t interf = NULL; /* the measured interferences */
- TBX_Graph_t builded_graph = NULL; /* the graph builded from the interferences */
-
- /* basics setups */
- if (argc>MAXHOSTS) {
- fprintf(stderr,"You gave more than %d sensors for this experiment. Increase the MAX HOSTS constant in alnem code to be bigger than this number.\n",argc);
- return 1;
- }
-
- if ((errcode=gras_sock_server_open(4000,5000,&(g->sock)))) {
- fprintf(stderr,"MAESTRO: Error %s encountered while opening the server socket\n",xbt_error_name(errcode));
- return 1;
- }
-
- if (grasbw_register_messages()) {
- gras_sock_close(g->sock);
- return 1;
- }
-
- for (a=1; a<argc; a++) {
- TBX_FIFO_insert(host_fifo,TBX_Graph_newNode(graph,argv[a], NULL));
- }
- TBX_Graph_fixNodeNumbering(graph);
- interf=TBX_Graph_newInterfTable(host_fifo);
- TBX_Graph_graphInterfTableInit(graph,interf);
-
- /* measure the bandwidths */
- begin=time(NULL);
- beginSim=gras_time();
- for (a=1; a<argc; a++) {
- for (b=1; b<argc; b++) {
- int test=0;
-
- if (a==b) continue;
- fprintf(stderr,"BW XP(%s %s)=",argv[a],argv[b]);
- while ((errcode=grasbw_request(argv[a],4000,argv[b],4000,bufSize,expSize,msgSize,
- &dummy,&(bw[a][b]))) && (errcode == timeout_error)) {
- test++;
- if (test==10) {
- fprintf(stderr,"MAESTRO: 10 Timeouts; giving up\n");
- return 1;
- }
- }
- if (errcode) {
- fprintf(stderr,"MAESTRO: Error %s encountered while doing the test\n",xbt_error_name(errcode));
- return 1;
- }
- fprintf(stderr,"%f Mb/s in %f sec\n",bw[a][b],dummy);
- }
- }
- fprintf(stderr,"Did all BW tests in %ld sec (%.2f simulated sec)\n",
- time(NULL)-begin,gras_time()-beginSim);
-
- /* saturation tests */
- for (a=1; a<argc; a++) {
- for (b=1; b<argc; b++) {
- for (c=1 ;c<argc; c++) {
- INTERF(graph,interf,argv[a],argv[c],argv[b],argv[c])= 1;
- }
- }
- }
-
- for (a=1; a<argc; a++) {
- for (b=1; b<argc; b++) {
- if (a==b) continue;
-
- if ((errcode=grasbw_saturate_start(argv[a],4000,argv[b],4000,satSize,360000000))) {
- fprintf(stderr,"MAESTRO: Error %s encountered while starting saturation\n",
- xbt_error_name(errcode));
- return -1;
- }
- gras_sleep(1,0);
-
- begin=time(NULL);
- beginSim=gras_time();
- for (c=1 ;c<argc; c++) {
- if (a==c || b==c) continue;
-
- for (d=1 ;d<argc; d++) {
- if (a==d || b==d || c==d) continue;
-
- if ((errcode=grasbw_request(argv[c],4000,argv[d],4000,bufSize,expSize,msgSize,
- &dummy,&(bw_sat[c][d])))) {
- fprintf(stderr,"MAESTRO: Error %s encountered in test\n",xbt_error_name(errcode));
- return 1;
- }
- fprintf(stderr, "MAESTRO[%.2f sec]: SATURATED BW XP(%s %s // %s %s) => %f (%f vs %f)%s\n",
- gras_time(),
- argv[c],argv[d],argv[a],argv[b],
- bw_sat[c][d]/bw[c][d],bw[c][d],bw_sat[c][d],
-
- (bw_sat[c][d]/bw[c][d] < 0.75) ? " THERE IS SOME INTERFERENCE !!!":"");
- INTERF(graph,interf,argv[c],argv[d],argv[a],argv[b])=
- (bw_sat[c][d]/bw[c][d] < 0.75) ? 1 : 0;
- }
- }
-
- if ((errcode=grasbw_saturate_stop(argv[a],4000,argv[b],4000))) {
- fprintf(stderr,"MAESTRO: Error %s encountered while stopping saturation\n",
- xbt_error_name(errcode));
- return -1;
- }
- fprintf(stderr,"Did an iteration on saturation pair in %ld sec (%.2f simulated sec)\n",
- time(NULL)-begin, gras_time()-beginSim);
- }
- }
-
- /* reconstruct the graph */
- TBX_Graph_interferenceTableDump(interf);
- TBX_Graph_interferenceTableSave(interf,"interference.dat");
- begin=time(NULL);
- fprintf(stderr, "MAESTRO: Reconstruct the graph... ");
- builded_graph = TBX_Graph_exploreInterference(interf);
- TBX_Graph_exportToGraphViz(builded_graph, "toto.dot");
- fprintf(stderr, "done (took %d sec)",(int)time(NULL));
-
- /* end */
- TBX_Graph_freeGraph(graph,NULL,NULL,NULL);
- TBX_Graph_freeGraph(builded_graph,NULL,NULL,NULL);
- TBX_Graph_freeInterfTable(interf);
-
- gras_sleep(5,0);
- exit(0); /* FIXME: There is a bug in MSG preventing me from terminating this server properly */
- return gras_sock_close(g->sock);
-}
+++ /dev/null
-/* $Id$ */
-
-/* ALNeM builder. Take an interference matrix as argument, */
-/* and reconstruct the corresponding graph itself */
-
-/* Copyright (c) 2003 Martin Quinson. 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 <stdio.h>
-#include <stdlib.h>
-
-#include <tbx_graph.h> /* alvin's graph toolbox (+ reconstruction algorithm) */
-
-int main(int argc,char *argv[]) {
- TBX_Graph_t graph; /* a dummy graph containing all hosts */
- TBX_FIFO_t host_fifo;
- TBX_InterfTable_t interf; /* the measured interferences */
- TBX_Graph_t builded_graph; /* the graph builded from the interferences */
-
- if (argc != 2) {
- fprintf(stderr,"alnem_builder: USAGE:\n");
- fprintf(stderr," alnem_builder interference_file\n");
- exit (1);
- }
-
- if (TBX_Graph_interferenceTableRead (argv[1],&graph,&interf,&host_fifo)) {
- fprintf(stderr,"Can't read the interference data, aborting\n");
- exit (1);
- }
-
- builded_graph = TBX_Graph_exploreInterference(interf);
- TBX_Graph_exportToGraphViz(builded_graph, "toto.dot");
- return 0;
-}
+++ /dev/null
-A sensor
-B sensor
-C sensor
-D sensor
-Master maestro A B C D
+++ /dev/null
-61 sensor
-62 sensor
-63 sensor
-69 sensor
-70 sensor
-77 sensor
-81 sensor
-83 sensor
-85 sensor
-87 sensor
-88 sensor
-95 sensor
-98 sensor
-107 sensor
-109 sensor
-111 sensor
-112 sensor
-121 sensor
-124 sensor
-125 sensor
-131 sensor
-145 sensor
-150 sensor
-156 sensor
-157 sensor
-162 sensor
-165 sensor
-168 sensor
-169 sensor
-170 sensor
-175 sensor
-177 sensor
-178 sensor
-109 maestro 61 62 63 69 70 77 81 83 85 98 107 109 111 112 121 124 125 131 145 150 156 157 162 165 168 169 170 175 177 178
+++ /dev/null
-INTERFERENCE GRAPH DUMPED TO FILE. DO NOT EDIT, EVEN TO ADD OR REMOVE A SPACE !!
-4
-A
-B
-C
-D
-1 1
-1 1
-1 0
-1 1
-1 0
-1 1
-1 1
-1 1
-1 0
-1 1
-1 0
-1 1
-1 1
-0 1
-1 1
-0 1
-1 1
-1 1
-1 1
-0 1
-1 1
-0 1
-1 1
-1 1
+++ /dev/null
-/* $Id$ */
-
-/* gras_rl - legacy implementation of GRAS on real life */
-/* This file should be KILLED whenever the raw sockets work in the new gras */
-
-/* Copyright (c) 2003 Martin Quinson. 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 "gras_rl.h"
-
-#include <stdlib.h>
-#include <string.h>
-
-#include <unistd.h> /* sleep() */
-
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <netinet/in.h> /* struct in_addr */
-#include <signal.h>
-
-
-/* NWS headers */
-#include "osutil.h"
-#include "timeouts.h"
-#include "protocol.h"
-
-/* XBT_LOG_NEW_DEFAULT_CATEGORY(rl,"Real Life"); Comment this to make sure it doesn't land into the documentation*/
-
-/* globals */
-static grasProcessData_t *_grasProcessData;
-
-/* Prototypes of internal functions */
-static int grasConversionRequired(const DataDescriptor *description, size_t howMany);
-static xbt_error_t
-_gras_rawsock_exchange(gras_rawsock_t *sd, int sender, unsigned int timeout,
- unsigned int expSize, unsigned int msgSize);
-
-
-xbt_error_t gras_process_init() {
- if (!(_grasProcessData=(grasProcessData_t *)malloc(sizeof(grasProcessData_t)))) {
- fprintf(stderr,"gras_process_init: cannot malloc %d bytes\n",sizeof(grasProcessData_t));
- return malloc_error;
- }
- _grasProcessData->grasMsgQueueLen=0;
- _grasProcessData->grasMsgQueue = NULL;
-
- _grasProcessData->grasCblListLen = 0;
- _grasProcessData->grasCblList = NULL;
-
- _grasProcessData->userdata = NULL;
- return no_error;
-}
-xbt_error_t gras_process_finalize() {
- fprintf(stderr,"FIXME: %s not implemented (=> leaking on exit :)\n",__FUNCTION__);
- return no_error;
-}
-
-/* **************************************************************************
- * Openning/Maintaining/Closing connexions
- * **************************************************************************/
-xbt_error_t
-gras_sock_client_open(const char *host, short port,
- /* OUT */ gras_sock_t **sock) {
-
- int addrCount;
- IPAddress addresses[10];
- int i;
- int sd;
-
- if (!(*sock=malloc(sizeof(gras_sock_t)))) {
- fprintf(stderr,"Malloc error\n");
- return malloc_error;
- }
- (*sock)->peer_addr=NULL;
-
- if (!(addrCount = IPAddressValues(host, addresses, 10))) {
- fprintf(stderr,"grasOpenClientSocket: address retrieval of '%s' failed\n",host);
- return system_error;
- }
-
- for(i = 0; i < addrCount && i<10 ; i++) {
- if(CallAddr(addresses[i], port, &sd, -1)) {
- (*sock)->sock = sd;
- (*sock)->port = port;
- return no_error;
- }
- }
- free(*sock);
- fprintf(stderr,"grasOpenClientSocket: something wicked happenned while connecting to %s:%d",
- host,port);
- return system_error;
-}
-
-
-xbt_error_t
-gras_sock_server_open(unsigned short startingPort, unsigned short endingPort,
- /* OUT */ gras_sock_t **sock) {
-
- unsigned short port;
-
- if (!(*sock=malloc(sizeof(gras_sock_t)))) {
- fprintf(stderr,"Malloc error\n");
- return malloc_error;
- }
-
- if (!EstablishAnEar(startingPort,endingPort,&((*sock)->sock),&port)) {
- free(*sock);
- return unknown_error;
- }
- (*sock)->peer_addr=NULL;
- (*sock)->port=port;
-
- return no_error;
-}
-
-xbt_error_t gras_sock_close(gras_sock_t *sock) {
- if (sock) {
- DROP_SOCKET(&(sock->sock));
- if (sock->peer_addr) free(sock->peer_addr);
- free(sock);
- }
- return no_error;
-}
-
-unsigned short
-gras_sock_get_my_port(gras_sock_t *sd) {
- if (!sd) return -1;
- return sd->port;
-}
-
-unsigned short
-gras_sock_get_peer_port(gras_sock_t *sd) {
- if (!sd) return -1;
- return PeerNamePort(sd->sock);
-}
-
-char *
-gras_sock_get_peer_name(gras_sock_t *sd) {
- char *tmp;
-
- if (!sd) return NULL;
- tmp=PeerName_r(sd->sock);
- if (tmp) {
- strcpy(sd->peer_name,tmp);
- free(tmp);
- } else {
- strcpy(sd->peer_name,"unknown");
- }
-
- return sd->peer_name;
-}
-
-
-/* **************************************************************************
- * format handling
- * **************************************************************************/
-
-/*
- * Returns 1 or 0 depending on whether or not format conversion is required for
- * data with the format described by the #howMany#-long array #description#.
- */
-int grasConversionRequired(const DataDescriptor *description, size_t howMany){
- int i;
-
- if(DataSize(description, howMany, HOST_FORMAT) !=
- DataSize(description, howMany, NETWORK_FORMAT)) {
- return 1;
- }
-
- for(i = 0; i < howMany; i++) {
- if(description[i].type == STRUCT_TYPE) {
- if(grasConversionRequired(description[i].members, description[i].length)) {
- return 1;
- }
- } else if(DifferentFormat(description[i].type))
- return 1;
- }
-
- return DifferentOrder();
-}
-
-/* **************************************************************************
- * Actually exchanging messages
- * **************************************************************************/
-
-/*
- * Discard data on the socket because of failure on our side
- */
-void
-gras_msg_discard(gras_sock_t *sd, size_t size) {
- char garbage[2048];
- int s=size;
-
- while(s > 0) {
- (void)RecvBytes(sd->sock,
- garbage,
- (s > sizeof(garbage)) ? sizeof(garbage) : s,
- GetTimeOut(RECV, Peer(sd->sock), 2048));
- s -= sizeof(garbage);
- }
-}
-
-int grasDataRecv( gras_sock_t *sd,
- void **data,
- const DataDescriptor *description,
- size_t description_length,
- unsigned int repetition) {
-
- void *converted=NULL;
- int convertIt;
- void *destination; /* where to receive the data from the net (*data or converted)*/
- int recvResult;
- size_t netSize,hostSize;
- double start; /* for timeouts */
- char *net,*host; /* iterators */
- int i;
-
- gras_lock();
-
- netSize = DataSize(description, description_length, NETWORK_FORMAT);
- hostSize = DataSize(description, description_length, HOST_FORMAT);
-
- if (!(*data=malloc(hostSize*repetition))) {
- gras_unlock();
- ERROR1("grasDataRecv: memory allocation of %d bytes failed\n", hostSize*repetition);
- return 0;
- }
-
- convertIt = grasConversionRequired(description, description_length);
-
- if(convertIt) {
- if (!(converted = malloc(netSize*repetition))) {
- free(*data);
- gras_unlock();
- ERROR1("RecvData: memory allocation of %d bytes failed\n", netSize*repetition);
- return 0;
- }
- destination = converted;
- } else {
- destination = *data;
- }
-
- /* adaptive timeout */
- start = CurrentTime();
-
- recvResult = RecvBytes(sd->sock, destination, netSize*repetition,
- GetTimeOut(RECV, Peer(sd->sock), netSize*repetition));
- /* we assume a failure is a timeout ... Shouldn't hurt
- * too much getting a bigger timeout anyway */
- SetTimeOut(RECV, sd->sock, CurrentTime()-start, netSize*repetition, !recvResult);
-
- fprintf(stderr,"RECV [seqLen=%d;netsize=%d;hostsize=%d] : (",
- repetition,netSize,hostSize);
- for (i=0; i<netSize * repetition; i++) {
- if (i) fputc('.',stderr);
- fprintf(stderr,"%02X",((unsigned char*)destination)[i]);
- }
- fprintf(stderr,") on %p.\n",destination);
-
- if (recvResult != 0) {
- if(convertIt) {
- for (i=0, net=(char*)converted, host=(char*)*data;
- i<repetition;
- i++, net += netSize, host += hostSize) {
- ConvertData((void*)host, (void*)net, description, description_length, NETWORK_FORMAT);
- }
- }
- if(converted != NULL)
- free(converted);
- }
- if (!gras_unlock()) return 0;
-
- return recvResult;
-}
-
-xbt_error_t grasDataSend(gras_sock_t *sd,
- const void *data,
- const DataDescriptor *description,
- size_t description_length,
- int repetition) {
-
- void *converted;
- char *net,*host; /* iterators */
- int sendResult,i;
- const void *source;
- size_t netSize = DataSize(description, description_length, NETWORK_FORMAT);
- size_t hostSize = DataSize(description, description_length, HOST_FORMAT);
- double start; /* for timeouts */
-
- gras_lock();
- converted = NULL;
-
- if(grasConversionRequired(description, description_length)) {
- converted = malloc(netSize * repetition);
- if(converted == NULL) {
- gras_unlock();
- fprintf(stderr,"grasDataSend: memory allocation of %d bytes failed.\n",netSize * repetition);
- return malloc_error;
- }
-
- for (i=0, net=(char*)converted, host=(char*)data;
- i<repetition;
- i++, net += netSize, host += hostSize)
- ConvertData((void*)net, (void*)host, description, description_length, HOST_FORMAT);
- source = converted;
- } else {
- source = data;
- }
-
- fprintf(stderr,"SEND (");
- for (i=0; i<netSize * repetition; i++) {
- if (i) fputc('.',stderr);
- fprintf(stderr,"%02X",((unsigned char*)source)[i]);
- }
- fprintf(stderr,") from %p\n",source);
- // grasDataDescDump((const DataDescriptor *)description, description_length);
-
- /* adaptive timeout */
- start = CurrentTime();
- sendResult = SendBytes(sd->sock, source, netSize * repetition,
- GetTimeOut(SEND, Peer(sd->sock),netSize*repetition));
- /* we assume a failure is a timeout ... Shouldn't hurt
- * too much getting a bigger timeout anyway */
- SetTimeOut(SEND, sd->sock, CurrentTime()-start, netSize * repetition, !sendResult);
- if(converted != NULL)
- free((void *)converted);
-
- gras_unlock();
- return no_error;
-}
-
-xbt_error_t
-grasMsgRecv(gras_msg_t **msg,
- double timeOut) {
- int dummyldap;
- gras_msg_t *res;
- size_t recvd; /* num of bytes received */
- int i;
- gras_sock_t *sd;
-
- if (!(sd = (gras_sock_t*)malloc(sizeof(gras_sock_t)))) {
- fprintf(stderr,"grasMsgRecv: Malloc error\n");
- return malloc_error;
- }
-
- if(!IncomingRequest(timeOut, &(sd->sock), &dummyldap)) {
- free(sd);
- return timeout_error;
- }
-
- if (!gras_lock()) {
- free (sd);
- return thread_error;
- }
- if (!(res = malloc(sizeof(gras_msg_t)))) {
- gras_unlock();
- *msg=NULL;
- free(sd);
- return malloc_error;
- }
- *msg=res;
- res->sock=sd;
- res->freeDirective=free_after_use;
-
- if (!gras_unlock()) return thread_error;
-
- if(!(recvd=grasDataRecv( sd,
- (void **)&(res->header),
- headerDescriptor,headerDescriptorCount,1))) {
- fprintf(stderr,"grasMsgRecv: no message received\n");
- return network_error;
- }
- res->header->dataSize -= recvd;
-
- fprintf(stderr,"Received header=ver:'%s' msg:%d size:%d seqCount:%d\n",
- res->header->version, res->header->message,
- res->header->dataSize, res->header->seqCount);
- if (strncmp(res->header->version,GRASVERSION,strlen(GRASVERSION))) {
- /* The other side do not use the same version than us. Let's panic */
- char junk[2046];
- int junkint;
- while ((junkint=recv(sd->sock,junk,2046,0)) == 2046);
- fprintf(stderr,"PANIC: Got a message from a peer (%s:%d) using GRAS version '%10s' while this side use version '%s'.\n",
- gras_sock_get_peer_name(sd),gras_sock_get_peer_port(sd),
- res->header->version,GRASVERSION);
- return mismatch_error;
- }
- if (!(res->entry=grasMsgEntryGet(res->header->message))) {
- /* This message is not registered on our side, discard it */
- gras_msg_discard(sd,res->header->dataSize);
-
- i= res->header->message;
- free(res->header);
- free(res);
- *msg=NULL;
-
- fprintf(stderr,"grasMsgRecv: unregistered message %d received from %s\n",
- i,gras_sock_get_peer_name(sd));
- return mismatch_error;
- }
-
- if (!(recvd=grasDataRecv( sd,
- (void **)&(res->dataCount),
- countDescriptor,res->entry->seqCount,1))) {
- gras_msg_discard(sd,res->header->dataSize);
- i = res->header->message;
- free(res->header);
- free(res);
- *msg=NULL;
- fprintf(stderr, "grasMsgRecv: Error while getting elemCounts in message %d received from %s\n",
- i,gras_sock_get_peer_name(sd));
- return network_error;
- }
- res->header->dataSize -= recvd;
-
- if (!gras_lock()) return thread_error;
- if (!(res->data=(void**)malloc(sizeof(void*)*res->entry->seqCount))) {
- gras_msg_discard(sd,res->header->dataSize);
-
- i= res->header->message;
- free(res->header);
- free(res->dataCount);
- free(res);
- *msg=NULL;
-
- gras_unlock();
- fprintf(stderr,"grasMsgRecv: Out of memory while receiving message %d received from %s\n",
- i,gras_sock_get_peer_name(sd));
- return malloc_error;
- }
- if (!gras_unlock()) return thread_error;
-
- for (i=0;i<res->entry->seqCount;i++) {
-
- if(!(recvd=grasDataRecv( sd,
- &(res->data[i]),
- (const DataDescriptor*)res->entry->dd[i],
- res->entry->ddCount[i],
- res->dataCount[i]) )) {
- gras_msg_discard(sd,res->header->dataSize);
- for (i--;i>=0;i--) free(res->data[i]);
- free(res->dataCount);
- free(res->data);
- free(res);
- *msg=NULL;
- fprintf(stderr,"grasDataRecv: Transmision error while receiving message %d received from %s\n",
- i,gras_sock_get_peer_name(sd));
- return network_error;
- }
- res->header->dataSize -= recvd;
- }
-
- if (res->header->dataSize) {
- fprintf(stderr,"Damnit dataSize = %d != 0 after receiving message %d received from %s\n",
- res->header->dataSize,i,gras_sock_get_peer_name(sd));
- return unknown_error;
- }
- return no_error;
-}
-
-/*
- * Send a message to the network
- */
-
-xbt_error_t
-gras_msg_send(gras_sock_t *sd,
- gras_msg_t *msg,
- e_xbt_free_directive_t freeDirective) {
-
- xbt_error_t errcode;
- int i;
-
- /* arg validity checks */
- xbt_assert0(msg,"Trying to send NULL message");
- xbt_assert0(sd, "Trying to send over a NULL socket");
-
-
- fprintf(stderr,"Header to send=ver:'%s' msg:%d size:%d seqCount:%d\n",
- msg->header->version, msg->header->message,
- msg->header->dataSize, msg->header->seqCount);
-
- /* Send the message */
- if ((errcode=grasDataSend(sd,
- msg->header,
- headerDescriptor,headerDescriptorCount,1))) {
- fprintf(stderr,"gras_msg_send: Error '%s' while sending header of message %s to %s:%d\n",
- xbt_error_name(errcode),msg->entry->name,gras_sock_get_peer_name(sd),gras_sock_get_peer_port(sd));
- return errcode;
- }
-
- if ((errcode=grasDataSend(sd,
- msg->dataCount,countDescriptor,countDescriptorCount,
- msg->entry->seqCount))) {
- fprintf(stderr,"gras_msg_send: Error '%s' while sending sequence counts of message %s to %s\n",
- xbt_error_name(errcode),msg->entry->name,gras_sock_get_peer_name(sd));
- return errcode;
- }
-
- for (i=0; i<msg->entry->seqCount; i++) {
- if ((errcode=grasDataSend(sd,
- msg->data[i],
- (const DataDescriptor*)msg->entry->dd[i],msg->entry->ddCount[i],
- msg->dataCount[i]))) {
- fprintf(stderr,"gras_msg_send: Error '%s' while sending sequence %d of message %s to %s\n",
- xbt_error_name(errcode),i,msg->entry->name,gras_sock_get_peer_name(sd));
- return errcode;
- }
- }
-
- if (freeDirective == free_after_use) gras_msg_free(msg);
- return no_error;
-}
-
-gras_sock_t *gras_sock_new(void) {
- return malloc(sizeof(gras_sock_t));
-}
-
-void grasSockFree(gras_sock_t *s) {
- if (s) free (s);
-}
-
-/* **************************************************************************
- * Creating/Using raw sockets
- * **************************************************************************/
-xbt_error_t gras_rawsock_server_open(unsigned short startingPort,
- unsigned short endingPort,
- unsigned int bufSize, gras_rawsock_t **sock) {
- struct sockaddr_in sockaddr;
-
-
- if (!(*sock=malloc(sizeof(gras_rawsock_t)))) {
- fprintf(stderr,"Malloc error\n");
- return malloc_error;
- }
-
-
- for((*sock)->port = startingPort;
- (*sock)->port <= endingPort;
- (*sock)->port++) {
- if(((*sock)->sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
- ERROR0("gras_rawsock_server_open: socket creation failed\n");
- free(*sock);
- return system_error;
- }
-
- setsockopt((*sock)->sock, SOL_SOCKET, SO_RCVBUF, (char *)&bufSize, sizeof(bufSize));
- setsockopt((*sock)->sock, SOL_SOCKET, SO_SNDBUF, (char *)&bufSize, sizeof(bufSize));
-
- memset(&sockaddr, 0, sizeof(sockaddr));
- sockaddr.sin_port = htons((unsigned short)(*sock)->port);
- sockaddr.sin_addr.s_addr = INADDR_ANY;
- sockaddr.sin_family = AF_INET;
-
- if (bind((*sock)->sock,
- (struct sockaddr *)&sockaddr, sizeof(sockaddr)) != -1 &&
- listen((*sock)->sock, 1) != -1) {
- break;
- }
- close((*sock)->sock);
- }
-
- if((*sock)->port > endingPort) {
- fprintf(stderr,"gras_rawsock_server_open: couldn't find a port between %d and %d\n",
- startingPort, endingPort);
- free(*sock);
- return mismatch_error;
- }
-
- return no_error;
-}
-
-void Dummy(int);
-void Dummy(int sig) {
- return;
-}
-
-xbt_error_t gras_rawsock_client_open(const char *host, short port,
- unsigned int bufSize, gras_rawsock_t **sock) {
- int i,addrCount;
- IPAddress addresses[10];
- void (*was)(int);
- struct sockaddr_in sockaddr;
-
- if (!(*sock=malloc(sizeof(gras_rawsock_t)))) {
- fprintf(stderr,"Malloc error\n");
- return malloc_error;
- }
- (*sock)->port=-1;
-
- if (!(addrCount = IPAddressValues(host, addresses, 10))) {
- fprintf(stderr,"grasOpenClientSocket: address retrieval of '%s' failed\n",host);
- free(*sock);
- return system_error;
- }
- (*sock)->port = port;
- memset(&sockaddr, 0, sizeof(sockaddr));
- sockaddr.sin_port = htons((unsigned short)(*sock)->port);
- sockaddr.sin_family = AF_INET;
-
- for(i = 0; i < addrCount && i<10 ; i++) {
- if(((*sock)->sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
- fprintf(stderr,"gras_rawsock_client_open: socket creation failed\n");
- free(*sock);
- return system_error;
- }
-
- setsockopt((*sock)->sock, SOL_SOCKET, SO_RCVBUF,
- (char *)&bufSize, sizeof(bufSize));
- setsockopt((*sock)->sock, SOL_SOCKET, SO_SNDBUF,
- (char *)&bufSize, sizeof(bufSize));
-
- sockaddr.sin_addr.s_addr = addresses[i];
-
- was = signal(SIGALRM,Dummy);
- SetRealTimer(GetTimeOut(CONN, addresses[i], 0));
- if(connect((*sock)->sock,
- (struct sockaddr *)&sockaddr, sizeof(sockaddr)) == 0) {
- SetRealTimer(0);
- signal(SIGALRM,was);
- break;
-
- }
- SetRealTimer(0);
- close((*sock)->sock);
- signal(SIGALRM,was);
- }
-
- if (i>=10) {
- free(*sock);
- fprintf(stderr,
- "grasOpenClientRawSocket: Unable to contact %s:%d\n",
- host,port);
- return network_error;
- }
-
- return no_error;
-}
-
-xbt_error_t gras_rawsock_close(gras_rawsock_t *sd) {
- if (sd) {
- CloseSocket(&(sd->sock), 0);
- free(sd);
- }
- return no_error;
-}
-
-unsigned short gras_rawsock_get_peer_port(gras_rawsock_t *sd) {
- if (!sd) return -1;
- return sd->port;
-}
-
-/* FIXME: RL ignores the provided timeout and compute an appropriate one */
-static xbt_error_t
-_gras_rawsock_exchange(gras_rawsock_t *sd, int sender, unsigned int timeout,
- unsigned int expSize, unsigned int msgSize){
- char *expData;
- int bytesThisCall;
- int bytesThisMessage;
- int bytesTotal;
-
- fd_set rd_set;
- int rv;
-
- char *name;
- IPAddress addr;
-
- int ltimeout;
- struct timeval timeOut;
-
- if((expData = (char *)malloc(msgSize)) == NULL) {
- fprintf(stderr,"gras_rawsock_send: malloc %d failed\n", msgSize);
- return malloc_error;
- }
-
- /* let's get information on the caller (needed later on) */
- name = PeerName_r(sd->sock);
- IPAddressValue(name, &addr);
- free(name);
-
- ltimeout = (int)GetTimeOut((sender ? SEND : RECV), addr, expSize/msgSize +1);
-
- if (sender)
- SetRealTimer(ltimeout);
-
-
- for(bytesTotal = 0;
- bytesTotal < expSize;
- bytesTotal += bytesThisMessage) {
- for(bytesThisMessage = 0;
- bytesThisMessage < msgSize;
- bytesThisMessage += bytesThisCall) {
-
-
- if(sender) {
- bytesThisCall = send(sd->sock, expData, msgSize - bytesThisMessage, 0);
- } else {
- bytesThisCall = 0;
- FD_ZERO(&rd_set);
- FD_SET(sd->sock,&rd_set);
- timeOut.tv_sec = ltimeout;
- timeOut.tv_usec = 0;
- /*
- * YUK! The timeout can get to be REALLY large if the
- * amount of data gets big -- fence it at 60 secs
- */
- if ((ltimeout <= 0) || (ltimeout > 60)) {
- ltimeout = 60;
- }
- rv = select(sd->sock+1,&rd_set,NULL,NULL,&timeOut);
- if(rv > 0) {
- bytesThisCall = recv(sd->sock, expData, msgSize-bytesThisMessage, 0);
- }
- }
- }
- }
- free(expData);
- return no_error;
-}
-
-xbt_error_t
-gras_rawsock_recv(gras_rawsock_t *sd, unsigned int expSize, unsigned int msgSize,
- unsigned int timeout) {
- return _gras_rawsock_exchange(sd,0,timeout,expSize,msgSize);
-}
-xbt_error_t
-gras_rawsock_send(gras_rawsock_t *sd, unsigned int expSize, unsigned int msgSize){
- return _gras_rawsock_exchange(sd,1,0,expSize,msgSize);
-}
-
-
-
-
-/* **************************************************************************
- * Process data
- * **************************************************************************/
-grasProcessData_t *grasProcessDataGet() {
- return _grasProcessData;
-}
-
-/* **************************************************************************
- * Wrappers over OS functions
- * **************************************************************************/
-double gras_time() {
- return MicroTime();
-}
-
-void gras_sleep(unsigned long sec,unsigned long usec) {
- sleep(sec);
- if (usec/1000000) sleep(usec/1000000);
- (void)usleep(usec % 1000000);
-}
+++ /dev/null
-/* $Id$ */
-
-/* gras_rl.h - private interface for legacy GRAS when on real life */
-/* This file should be KILLED whenever the raw sockets work in the new gras */
-
-/* Copyright (c) 2003 Martin Quinson. 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 GRAS_RL_H
-#define GRAS_RL_H
-
-#ifdef GRAS_SG_H
-#error Impossible to load gras_sg.h and gras_rl.h at the same time
-#endif
-
-#include "gras_private.h"
-//#include "diagnostic.h"
-
-BEGIN_DECL
-
-/****************************************************************************/
-/****************************************************************************/
-/* Openning/Maintaining/Closing connexions */
-/****************************************************************************/
-/****************************************************************************/
-
-/* Declaration of the socket type */
-#define PEER_NAME_LEN 200
-struct gras_sock_s {
- int sock;
- int port; /* port on this side */
- char peer_name[PEER_NAME_LEN]; /* buffer to use PeerName_r */
- char *peer_addr;
-};
-
-struct gras_rawsock_s {
- int sock;
- int port;
-};
-
-/****************************************************************************/
-/****************************************************************************/
-/* Format convertion stuff (copied verbatim from formatutil.h because I need*/
-/* to put datadescriptor in gras.h and the functions only in RL) */
-/****************************************************************************/
-/****************************************************************************/
-
-void ConvertData(void *destination,
- const void *source,
- const DataDescriptor *description,
- size_t length,
- FormatTypes sourceFormat);
-
-
-int DifferentFormat(DataTypes whatType);
-int DifferentOrder(void);
-int DifferentSize(DataTypes whatType);
-void HomogenousConvertData(void *destination,
- const void *source,
- DataTypes whatType,
- size_t repetitions,
- FormatTypes sourceFormat);
-size_t HomogenousDataSize(DataTypes whatType,
- size_t repetitions,
- FormatTypes format);
-void ReverseData(void *destination,
- const void *source,
- DataTypes whatType,
- int repetitions,
- FormatTypes format);
-
-
-/****************************************************************************/
-/****************************************************************************/
-/* Messaging stuff (reimplementation of message.h) */
-/****************************************************************************/
-/****************************************************************************/
-/**
- * gras_msg_discard:
- * @sd: Socket on which this message arrives
- * @size: Size of the message to discard.
- *
- * Discard data on the socket because of failure on our side (out of mem or msg
- * type unknown).
- */
-void
-gras_msg_discard(gras_sock_t *sd, size_t size);
-
-/**
- * grasRecvData:
- * @sd: Socket to listen on
- * @data: Where to store the data (allocated by this function)
- * @description: What data to expect.
- * @Returns: the number of bytes read for that.
- *
- * Receive data from the network on a buffer allocated by this function for that
- */
-int
-grasDataRecv( gras_sock_t *sd,
- void **data,
- const DataDescriptor *description,
- size_t description_length,
- unsigned int repetition);
-
-/**
- * grasRecvData:
- * @sd: Socket to write on
- * @data: What to send
- * @description: Format of the data
- * @Returns: the number of bytes written
- *
- * Send a sequence of data across the network
- */
-xbt_error_t
-grasDataSend(gras_sock_t *sd,
- const void *data,
- const DataDescriptor *description,
- size_t description_length,
- int repetition);
-
-END_DECL
-
-#endif /* GRAS_SG_H */
+++ /dev/null
-/* $Id$ */
-
-/* gras_sg - legacy implementation of GRAS on top of the SimGrid simulator */
-/* This file should be KILLED whenever the raw sockets work in the new gras */
-
-/* Copyright (c) 2003 Martin Quinson. 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 <stdio.h>
-#include <string.h>
-
-#include "gras_sg.h"
-
-XBT_LOG_DEFAULT_CATEGORY(GRAS);
-
-/* **************************************************************************
- * Openning/Maintaining/Closing connexions (private functions for both raw
- * and regular sockets)
- * **************************************************************************/
-xbt_error_t
-_gras_sock_server_open(unsigned short startingPort, unsigned short endingPort,
- int raw, unsigned int bufSize, /* OUT */ gras_sock_t **sock) {
-
- gras_hostdata_t *hd=hd=(gras_hostdata_t *)MSG_host_get_data(MSG_host_self());
- grasProcessData_t *pd=(grasProcessData_t *)MSG_process_get_data(MSG_process_self());
- int port,i;
- const char *host=MSG_host_get_name(MSG_host_self());
-
- xbt_assert0(hd,"HostData=NULL !! Please run grasInit on each process\n");
- xbt_assert0(pd,"ProcessData=NULL !! Please run grasInit on each process\n");
-
- for (port=startingPort ; port <= endingPort ; port++) {
- for (i=0; i<hd->portLen && hd->port[i] != port; i++);
- if (i<hd->portLen && hd->port[i] == port)
- continue;
-
- /* port not used so far. Do it */
- if (i == hd->portLen) {
- /* need to enlarge the tables */
- if (hd->portLen++) {
- hd->port2chan=(int*)realloc(hd->port2chan,hd->portLen*sizeof(int));
- hd->port =(int*)realloc(hd->port ,hd->portLen*sizeof(int));
- hd->raw =(int*)realloc(hd->raw ,hd->portLen*sizeof(int));
- } else {
- hd->port2chan=(int*)malloc(hd->portLen*sizeof(int));
- hd->port =(int*)malloc(hd->portLen*sizeof(int));
- hd->raw =(int*)malloc(hd->portLen*sizeof(int));
- }
- if (!hd->port2chan || !hd->port || !hd->raw) {
- fprintf(stderr,"GRAS: PANIC: A malloc error did lose all ports attribution on this host\n");
- hd->portLen = 0;
- return malloc_error;
- }
- }
- hd->port2chan[ i ]=raw ? pd->rawChan : pd->chan;
- hd->port[ i ]=port;
- hd->raw[ i ]=raw;
-
- /* Create the socket */
- if (!(*sock=(gras_sock_t*)malloc(sizeof(gras_sock_t)))) {
- fprintf(stderr,"GRAS: openServerSocket: out of memory\n");
- return malloc_error;
- }
-
- (*sock)->server_sock = 1;
- (*sock)->raw_sock = raw;
- (*sock)->from_PID = -1;
- (*sock)->to_PID = MSG_process_self_PID();
- (*sock)->to_host = MSG_host_self();
- (*sock)->to_port = port;
- (*sock)->to_chan = pd->chan;
-
- /*
- fprintf(stderr,"GRAS: '%s' (%d) ears on %s:%d%s (%p).\n",
- MSG_process_get_name(MSG_process_self()), MSG_process_self_PID(),
- host,port,raw? " (mode RAW)":"",*sock);
- */
- return no_error;
- }
- /* if we go out of the previous for loop, that's that we didn't find any
- suited port number */
-
- fprintf(stderr,
- "GRAS: can't find an empty port between %d and %d to open a socket on host %s\n.",
- startingPort,endingPort,host);
- return mismatch_error;
-}
-
-xbt_error_t
-_gras_sock_client_open(const char *host, short port, int raw, unsigned int bufSize,
- /* OUT */ gras_sock_t **sock) {
- m_host_t peer;
- gras_hostdata_t *hd;
- int i;
-
- /* make sure this socket will reach someone */
- if (!(peer=MSG_get_host_by_name(host))) {
- fprintf(stderr,"GRAS: can't connect to %s: no such host.\n",host);
- return mismatch_error;
- }
- if (!(hd=(gras_hostdata_t *)MSG_host_get_data(peer))) {
- fprintf(stderr,"GRAS: can't connect to %s: no process on this host.\n",host);
- return mismatch_error;
- }
- for (i=0; i<hd->portLen && port != hd->port[i]; i++);
- if (i == hd->portLen) {
- fprintf(stderr,"GRAS: can't connect to %s:%d, no process listen on this port.\n",host,port);
- return mismatch_error;
- }
-
- if (hd->raw[i] && !raw) {
- fprintf(stderr,"GRAS: can't connect to %s:%d in regular mode, the process listen in raw mode on this port.\n",host,port);
- return mismatch_error;
- }
- if (!hd->raw[i] && raw) {
- fprintf(stderr,"GRAS: can't connect to %s:%d in raw mode, the process listen in regular mode on this port.\n",host,port);
- return mismatch_error;
- }
-
-
- /* Create the socket */
- if (!(*sock=(gras_sock_t*)malloc(sizeof(gras_sock_t)))) {
- fprintf(stderr,"GRAS: openClientSocket: out of memory\n");
- return malloc_error;
- }
-
- (*sock)->server_sock = 0;
- (*sock)->raw_sock = raw;
- (*sock)->from_PID = MSG_process_self_PID();
- (*sock)->to_PID = hd->proc[ hd->port2chan[i] ];
- (*sock)->to_host = peer;
- (*sock)->to_port = port;
- (*sock)->to_chan = hd->port2chan[i];
-
- /*
- fprintf(stderr,"GRAS: %s (PID %d) connects in %s mode to %s:%d (to_PID=%d).\n",
- MSG_process_get_name(MSG_process_self()), MSG_process_self_PID(),
- raw?"RAW":"regular",host,port,(*sock)->to_PID);
- */
- return no_error;
-}
-
-xbt_error_t _gras_sock_close(int raw, gras_sock_t *sd) {
- gras_hostdata_t *hd=(gras_hostdata_t *)MSG_host_get_data(MSG_host_self());
- int i;
-
- xbt_assert0(hd,"HostData=NULL !! Please run grasInit on each process\n");
-
- if (!sd) return no_error;
- if (raw && !sd->raw_sock) {
- fprintf(stderr,"GRAS: gras_rawsock_close: Was passed a regular socket. Please use gras_sock_close()\n");
- }
- if (!raw && sd->raw_sock) {
- fprintf(stderr,"GRAS: grasSockClose: Was passed a raw socket. Please use gras_rawsock_close()\n");
- }
- if (sd->server_sock) {
- /* server mode socket. Un register it from 'OS' tables */
- for (i=0;
- i<hd->portLen && sd->to_port != hd->port[i];
- i++);
-
- if (i==hd->portLen) {
- fprintf(stderr,"GRAS: closeSocket: The host does not know this server socket.\n");
- } else {
- memmove(&(hd->port[i]), &(hd->port[i+1]), (hd->portLen -i -1) * sizeof(int));
- memmove(&(hd->raw[i]), &(hd->raw[i+1]), (hd->portLen -i -1) * sizeof(int));
- memmove(&(hd->port2chan[i]), &(hd->port2chan[i+1]), (hd->portLen -i -1) * sizeof(int));
- hd->portLen--;
- }
- }
- free(sd);
- return no_error;
-}
-
-/* **************************************************************************
- * Creating/Using regular sockets
- * **************************************************************************/
-xbt_error_t
-gras_sock_server_open(unsigned short startingPort, unsigned short endingPort,
- /* OUT */ gras_sock_t **sock) {
- return _gras_sock_server_open(startingPort,endingPort,0,0,sock);
-}
-
-xbt_error_t
-gras_sock_client_open(const char *host, short port,
- /* OUT */ gras_sock_t **sock) {
- return _gras_sock_client_open(host,port,0,0,sock);
-}
-
-xbt_error_t gras_sock_close(gras_sock_t *sd) {
- return _gras_sock_close(0,sd);
-}
-
-unsigned short
-gras_sock_get_my_port(gras_sock_t *sd) {
- if (!sd || !sd->server_sock) return -1;
- return sd->to_port;
-}
-
-unsigned short
-gras_sock_get_peer_port(gras_sock_t *sd) {
- if (!sd || sd->server_sock) return -1;
- return sd->to_port;
-}
-
-char *
-gras_sock_get_peer_name(gras_sock_t *sd) {
- m_process_t proc;
-
- if (!sd) return NULL;
- if ((proc=MSG_process_from_PID(sd->to_PID))) {
- return (char*) MSG_host_get_name(MSG_process_get_host(proc));
- } else {
- fprintf(stderr,"GRAS: try to access hostname of unknown process %d\n",sd->to_PID);
- return (char*) "(dead or unknown host)";
- }
-}
-/* **************************************************************************
- * Creating/Using raw sockets
- * **************************************************************************/
-xbt_error_t gras_rawsock_server_open(unsigned short startingPort, unsigned short endingPort,
- unsigned int bufSize, gras_rawsock_t **sock) {
- return _gras_sock_server_open(startingPort,endingPort,1,bufSize,(gras_sock_t**)sock);
-}
-
-xbt_error_t gras_rawsock_client_open(const char *host, short port,
- unsigned int bufSize, gras_rawsock_t **sock) {
- return _gras_sock_client_open(host,port,1,bufSize,(gras_sock_t**)sock);
-}
-
-xbt_error_t gras_rawsock_close(gras_rawsock_t *sd) {
- return _gras_sock_close(1,(gras_sock_t*)sd);
-}
-
-unsigned short
-gras_rawsock_get_peer_port(gras_rawsock_t *sd) {
- if (!sd || !sd->server_sock) return -1;
- return sd->to_port;
-}
-
-xbt_error_t
-gras_rawsock_send(gras_rawsock_t *sock, unsigned int expSize, unsigned int msgSize) {
- unsigned int bytesTotal;
- static unsigned int count=0;
- m_task_t task=NULL;
- char name[256];
-
- xbt_assert0(sock->raw_sock,"Asked to send raw data on a regular socket\n");
-
- for(bytesTotal = 0;
- bytesTotal < expSize;
- bytesTotal += msgSize) {
-
- sprintf(name,"Raw data[%d]",count++);
-
- task=MSG_task_create(name,0,((double)msgSize)/(1024.0*1024.0),NULL);
- /*
- fprintf(stderr, "%f:%s: gras_rawsock_send(%f %s -> %s) BEGIN\n",
- gras_time(),
- MSG_process_get_name(MSG_process_self()),
- ((double)msgSize)/(1024.0*1024.0),
- MSG_host_get_name( MSG_host_self()),
- MSG_host_get_name( sock->to_host));
- */
- if (MSG_task_put(task, sock->to_host,sock->to_chan) != MSG_OK) {
- fprintf(stderr,"GRAS: msgSend: Problem during the MSG_task_put()\n");
- return unknown_error;
- }
- /*fprintf(stderr, "%f:%s: gras_rawsock_send(%f -> %s) END\n",
- gras_time(),
- MSG_process_get_name(MSG_process_self()),
- ((double)msgSize)/(1024.0*1024.0),
- MSG_host_get_name( sock->to_host));*/
- }
- return no_error;
-}
-
-xbt_error_t
-gras_rawsock_recv(gras_rawsock_t *sock, unsigned int expSize, unsigned int msgSize,
- unsigned int timeout) {
- grasProcessData_t *pd=(grasProcessData_t *)MSG_process_get_data(MSG_process_self());
- unsigned int bytesTotal;
- m_task_t task=NULL;
- double startTime;
-
- xbt_assert0(sock->raw_sock,"Asked to receive raw data on a regular socket\n");
-
- startTime=gras_time();
- for(bytesTotal = 0;
- bytesTotal < expSize;
- bytesTotal += msgSize) {
-
- task=NULL;
- /*
- fprintf(stderr, "%f:%s: gras_rawsock_recv() BEGIN\n",
- gras_time(),
- MSG_process_get_name(MSG_process_self()));
- */
- do {
- if (MSG_task_Iprobe((m_channel_t) pd->rawChan)) {
- if (MSG_task_get(&task, (m_channel_t) pd->rawChan) != MSG_OK) {
- fprintf(stderr,"GRAS: Error in MSG_task_get()\n");
- return unknown_error;
- }
- if (MSG_task_destroy(task) != MSG_OK) {
- fprintf(stderr,"GRAS: Error in MSG_task_destroy()\n");
- return unknown_error;
- }
- /*
- fprintf(stderr, "%f:%s: gras_rawsock_recv() END\n",
- gras_time(),
- MSG_process_get_name(MSG_process_self()));
- */
- break;
- } else {
- MSG_process_sleep(0.0001);
- }
- } while (gras_time() - startTime < timeout);
-
- if (gras_time() - startTime > timeout)
- return timeout_error;
- }
- return no_error;
-}
-
-
-/* **************************************************************************
- * Actually exchanging messages
- * **************************************************************************/
-
-xbt_error_t
-grasMsgRecv(gras_msg_t **msg,
- double timeOut) {
-
- double startTime=gras_time();
- grasProcessData_t *pd=grasProcessDataGet();
- m_task_t task=NULL;
-
- do {
- if (MSG_task_Iprobe((m_channel_t) pd->chan)) {
- if (MSG_task_get(&task, (m_channel_t) pd->chan) != MSG_OK) {
- fprintf(stderr,"GRAS: Error in MSG_task_get()\n");
- return unknown_error;
- }
-
- *msg=(gras_msg_t*)MSG_task_get_data(task);
- /*
- {
- int i,j;
- gras_msg_t *__dm_=*msg;
-
- fprintf(stderr, "grasMsgRecv(%s) = %d seq (",
- __dm_->entry->name, __dm_->entry->seqCount );
-
- for (i=0; i<__dm_->entry->seqCount; i++) {
- fprintf(stderr,"%d elem {",__dm_->dataCount[i]);
- for (j=0; j<__dm_->dataCount[i]; j++) {
- fprintf(stderr,"%p; ",__dm_->data[i]);
- }
- fprintf(stderr,"},");
- }
- fprintf(stderr, ")\n");
- }
- */
-
- if (MSG_task_destroy(task) != MSG_OK) {
- fprintf(stderr,"GRAS: Error in MSG_task_destroy()\n");
- return unknown_error;
- }
- return no_error;
-
- } else {
- MSG_process_sleep(1);
- }
- } while (gras_time()-startTime < timeOut || MSG_task_Iprobe((m_channel_t) pd->chan));
- return timeout_error;
-}
-
-xbt_error_t
-gras_msg_send(gras_sock_t *sd,
- gras_msg_t *_msg,
- e_xbt_free_directive_t freeDirective) {
-
- grasProcessData_t *pd=grasProcessDataGet();
- m_task_t task;
- static int count=0;
- char name[256];
- gras_msg_t *msg;
- gras_sock_t *answer;
-
- /* arg validity checks */
- xbt_assert0(msg,"Trying to send NULL message");
- xbt_assert0(sd ,"Trying to send over a NULL socket");
-
- if (!(answer=(gras_sock_t*)malloc(sizeof(gras_sock_t)))) {
- RAISE_MALLOC;
- }
- answer->server_sock=0;
- answer->raw_sock =0;
- answer->from_PID = sd->to_PID;
- answer->to_PID = MSG_process_self_PID();
- answer->to_host = MSG_host_self();
- answer->to_port = 0;
- answer->to_chan = pd->chan;
-
- sprintf(name,"%s[%d]",_msg->entry->name,count++);
- /* if freeDirective == free_never, we have to build a copy of the message */
- if (freeDirective == free_never) {
- msg=gras_msg_copy(_msg);
- } else {
- msg=_msg;
- }
- msg->sock = answer;
-
- /*
- fprintf(stderr,"Send %s with answer(%p)=",msg->entry->name,msg->sock);
- fprintf(stderr,"(server=%d,raw=%d,fromPID=%d,toPID=%d,toHost=%p,toPort=%d,toChan=%d)\n",
- msg->sock->server_sock,msg->sock->raw_sock,msg->sock->from_PID,
- msg->sock->to_PID,msg->sock->to_host,msg->sock->to_port,msg->sock->to_chan);
- fprintf(stderr,"Send over %p=(server=%d,raw=%d,fromPID=%d,toPID=%d,toHost=%p,toPort=%d,toChan=%d)\n",
- sd,sd->server_sock,sd->raw_sock,sd->from_PID,sd->to_PID,sd->to_host,sd->to_port,sd->to_chan);
- */
-
- /*
- {
- int i,j;
- gras_msg_t *__dm_=msg;
-
- fprintf(stderr, "gras_msg_send(%s) = %d seq (",
- __dm_->entry->name, __dm_->entry->seqCount );
-
- for (i=0; i<__dm_->entry->seqCount; i++) {
- fprintf(stderr,"%d elem {",__dm_->dataCount[i]);
- for (j=0; j<__dm_->dataCount[i]; j++) {
- fprintf(stderr,"%p; ",__dm_->data[i]);
- }
- fprintf(stderr,"},");
- }
- fprintf(stderr, ")\n");
- }
- */
-
- /* Send it */
- task=MSG_task_create(name,0,((double)msg->header->dataSize)/(1024.0*1024.0),msg);
- if (MSG_task_put(task, sd->to_host,sd->to_chan) != MSG_OK) {
- fprintf(stderr,"GRAS: msgSend: Problem during the MSG_task_put()\n");
- return unknown_error;
- }
- return no_error;
-}
-
-gras_sock_t *gras_sock_new(void) {
- return malloc(sizeof(gras_sock_t));
-}
-
-void grasSockFree(gras_sock_t *s) {
- if (s) free (s);
-}
-
-/* **************************************************************************
- * Process data
- * **************************************************************************/
-grasProcessData_t *grasProcessDataGet() {
- return (grasProcessData_t *)MSG_process_get_data(MSG_process_self());
-}
-
-/* **************************************************************************
- * Wrappers over OS functions
- * **************************************************************************/
-double gras_time() {
- return MSG_getClock();
-}
-
-void gras_sleep(unsigned long sec, unsigned long usec) {
- MSG_process_sleep((double)sec + ((double)usec)/1000000);
-}
-
