Mercurial > cgi-bin > hgwebdir.cgi > VMS > C_Libraries > BestEffortMessaging

annotate LossyCom.c @ 2:b6dd31dbab8c

find changesets by author, revision, files, or words in the commit message
Receiving only messages for the current endpoint
author Merten Sach <msach@mailbox.tu-berlin.de>
date Mon, 12 Mar 2012 18:35:10 +0100
parents 826448e34e80
children 5c0fb7c519d7
rev   line source
msach@0 1
msach@0 2 /*
msach@0 3 * For a detailed description see header file.
msach@0 4 */
msach@0 5
msach@0 6 #include "LossyCom.h"
msach@0 7
msach@0 8 #include "VMS_Implementations/VMS_impl/vmalloc.h"
msach@0 9
msach@0 10 /*
msach@0 11 * Initializes the central exchange structure.
msach@0 12 * Allocates memory to fit the number of endpoints.
msach@0 13 * Returns NULL if an error occurs.
msach@0 14 */
msach@0 15 lossyCom__exchange_t* lossyCom__initialize(uint16_t numEndpoints)
msach@0 16 {
msach@0 17 lossyCom__exchange_t* exchange;
msach@0 18
msach@0 19 exchange = VMS_WL__malloc(sizeof(lossyCom__exchange_t));
msach@0 20 if(exchange == NULL)
msach@0 21 return NULL;
msach@0 22
msach@0 23 exchange->triggerCounter = 0;
msach@0 24 exchange->numEndpoints = numEndpoints;
msach@0 25 exchange->outboxArray = VMS_WL__malloc(sizeof(lossyCom__msg_t)*numEndpoints);
msach@0 26 if(exchange->outboxArray == NULL){
msach@0 27 VMS_WL__free(exchange);
msach@0 28 return NULL;
msach@0 29 }
msach@0 30
msach@0 31 return exchange;
msach@0 32 }
msach@0 33
msach@1 34 /*
msach@1 35 * Connects the local endpoint to the central exchange structure.
msach@1 36 * This registers a message handler that handles all the incomming messages.
msach@1 37 * Also an endpointID is set this ID has to be between 0 and total number of
msach@1 38 * endpoints-1 and the number has to be unique.
msach@1 39 */
msach@0 40 void lossyCom__initialize_endpoint(lossyCom__endpoint_t* localEndpoint,
msach@0 41 lossyCom__exchange_t* centralExchange,
msach@0 42 lossyCom__endpointID_t endpointID,
msach@1 43 lossyCom__msgHandler msgHandler,
msach@1 44 void* msgHandlerData)
msach@0 45 {
msach@0 46 localEndpoint->localTriggerCopy = 0;
msach@0 47 localEndpoint->endpointID = endpointID;
msach@1 48 localEndpoint->centralExchange = centralExchange;
msach@1 49 localEndpoint->msgHandler = msgHandler;
msach@1 50 localEndpoint->msgHandlerData = msgHandlerData;
msach@0 51 }
msach@0 52
msach@1 53 /*
msach@1 54 * This broadcasts a message to all connected receivers
msach@1 55 */
msach@0 56 void inline lossyCom__broadcastMsg(lossyCom__endpoint_t* localEndpoint,
msach@0 57 lossyCom__msgBody_t msg)
msach@0 58 {
msach@0 59 lossyCom__sendMsg(localEndpoint,
msach@0 60 BROADCAST_ID,
msach@0 61 msg);
msach@0 62 }
msach@0 63
msach@1 64 /*
msach@1 65 * This sends a message another endpoint. Again it is not guaranteed that the
msach@1 66 * message is received. But in most cases it will.
msach@1 67 */
msach@0 68 void inline lossyCom__sendMsg(lossyCom__endpoint_t* localEndpoint,
msach@0 69 lossyCom__endpointID_t receiverEndpointID,
msach@0 70 lossyCom__msgBody_t msg)
msach@0 71 {
msach@0 72 lossyCom__msg_t msgDraft;
msach@0 73 uint16_t triggerCopy;
msach@0 74
msach@1 75 //build message
msach@0 76 msgDraft = (0 | msg);
msach@0 77 msgDraft |= ((lossyCom__msg_t)receiverEndpointID << ENDPOINT_ID_SHIFT);
msach@0 78
msach@1 79 triggerCopy = localEndpoint->centralExchange->triggerCounter +1;
msach@0 80 msgDraft |= ((lossyCom__msg_t)triggerCopy << TRIGGER_SHIFT);
msach@0 81
msach@0 82 //write msg to central exchange
msach@1 83 localEndpoint->centralExchange->outboxArray[localEndpoint->endpointID] =
msach@0 84 msgDraft;
msach@0 85
msach@1 86 localEndpoint->centralExchange->triggerCounter = triggerCopy;
msach@2 87 //printf("send updated trigger to %d\n", triggerCopy);
msach@0 88 }
msach@0 89
msach@0 90 void inline lossyCom__receiveMsg(lossyCom__endpoint_t* localEndpoint)
msach@0 91 {
msach@1 92 uint16_t currentTriggerCopy;
msach@0 93 lossyCom__endpointID_t senderEndpointID;
msach@2 94 lossyCom__endpointID_t receiverID;
msach@0 95 lossyCom__msg_t msgCopy;
msach@0 96 uint16_t msgTrigger;
msach@0 97 lossyCom__msgBody_t msgBody;
msach@0 98
msach@0 99 senderEndpointID = 0;
msach@1 100 currentTriggerCopy = localEndpoint->centralExchange->triggerCounter;
msach@1 101
msach@0 102 //new message arrived if trigger counter is higher than the last time read
msach@2 103 if(currentTriggerCopy > localEndpoint->localTriggerCopy)
msach@0 104 {
msach@2 105 while(senderEndpointID < localEndpoint->centralExchange->numEndpoints)
msach@0 106 {
msach@2 107 if(senderEndpointID != localEndpoint->endpointID)
msach@0 108 {
msach@2 109 msgCopy = localEndpoint->centralExchange->outboxArray[senderEndpointID];
msach@2 110 msgTrigger = 0xFFFF & (msgCopy >> TRIGGER_SHIFT);
msach@2 111 // check if the message is new (msg trigger > archived trigger)
msach@2 112 // and already valid (msgTrigger <= currentTriggerCopy)
msach@2 113 if(msgTrigger > localEndpoint->localTriggerCopy &&
msach@2 114 msgTrigger <= currentTriggerCopy)
msach@2 115 {
msach@2 116 //printf("receive search until %d \n", currentTriggerCopy);
msach@2 117 //let the message handler parse the message
msach@2 118 msgBody = 0xFFFFFFFF & msgCopy;
msach@2 119 receiverID = 0xFFFF & (msgCopy >> ENDPOINT_ID_SHIFT);
msach@2 120 //only receive broadcast and p2p for own receiverID
msach@2 121 if(receiverID == BROADCAST_ID ||
msach@2 122 receiverID == localEndpoint->endpointID)
msach@2 123 {
msach@2 124 (*(localEndpoint->msgHandler))(senderEndpointID,
msach@2 125 msgBody,
msach@2 126 localEndpoint->msgHandlerData);
msach@2 127 }
msach@2 128 }
msach@0 129 }
msach@2 130 senderEndpointID++;
msach@0 131 }
msach@0 132 }
msach@2 133 //save last TriggerCounter of last parsed Msg
msach@1 134 localEndpoint->localTriggerCopy = currentTriggerCopy;
msach@0 135 }