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

comparison LossyCom.c @ 1:826448e34e80

find changesets by author, revision, files, or words in the commit message
added data to message handler
author Merten Sach <msach@mailbox.tu-berlin.de>
date Thu, 08 Mar 2012 20:06:00 +0100
parents 29d8b41926f0
children b6dd31dbab8c
comparison
equal deleted inserted replaced
0:c1c6692f2648 1:618d1c003877
29 } 29 }
30 30
31 return exchange; 31 return exchange;
32 } 32 }
33 33
34 /*
35 * Connects the local endpoint to the central exchange structure.
36 * This registers a message handler that handles all the incomming messages.
37 * Also an endpointID is set this ID has to be between 0 and total number of
38 * endpoints-1 and the number has to be unique.
39 */
34 void lossyCom__initialize_endpoint(lossyCom__endpoint_t* localEndpoint, 40 void lossyCom__initialize_endpoint(lossyCom__endpoint_t* localEndpoint,
35 lossyCom__exchange_t* centralExchange, 41 lossyCom__exchange_t* centralExchange,
36 lossyCom__endpointID_t endpointID, 42 lossyCom__endpointID_t endpointID,
37 lossyCom__msgHandler* msgHandler) 43 lossyCom__msgHandler msgHandler,
44 void* msgHandlerData)
38 { 45 {
39 localEndpoint->localTriggerCopy = 0; 46 localEndpoint->localTriggerCopy = 0;
40 localEndpoint->endpointID = endpointID; 47 localEndpoint->endpointID = endpointID;
41 localEndpoint->centralExchangePtr = centralExchange; 48 localEndpoint->centralExchange = centralExchange;
42 localEndpoint->msgHandler = msgHandler; 49 localEndpoint->msgHandler = msgHandler;
50 localEndpoint->msgHandlerData = msgHandlerData;
43 } 51 }
44 52
53 /*
54 * This broadcasts a message to all connected receivers
55 */
45 void inline lossyCom__broadcastMsg(lossyCom__endpoint_t* localEndpoint, 56 void inline lossyCom__broadcastMsg(lossyCom__endpoint_t* localEndpoint,
46 lossyCom__msgBody_t msg) 57 lossyCom__msgBody_t msg)
47 { 58 {
48 lossyCom__sendMsg(localEndpoint, 59 lossyCom__sendMsg(localEndpoint,
49 BROADCAST_ID, 60 BROADCAST_ID,
50 msg); 61 msg);
51 } 62 }
52 63
64 /*
65 * This sends a message another endpoint. Again it is not guaranteed that the
66 * message is received. But in most cases it will.
67 */
53 void inline lossyCom__sendMsg(lossyCom__endpoint_t* localEndpoint, 68 void inline lossyCom__sendMsg(lossyCom__endpoint_t* localEndpoint,
54 lossyCom__endpointID_t receiverEndpointID, 69 lossyCom__endpointID_t receiverEndpointID,
55 lossyCom__msgBody_t msg) 70 lossyCom__msgBody_t msg)
56 { 71 {
57 lossyCom__msg_t msgDraft; 72 lossyCom__msg_t msgDraft;
58 uint16_t triggerCopy; 73 uint16_t triggerCopy;
59 74
75 //build message
60 msgDraft = (0 | msg); 76 msgDraft = (0 | msg);
61 msgDraft |= ((lossyCom__msg_t)receiverEndpointID << ENDPOINT_ID_SHIFT); 77 msgDraft |= ((lossyCom__msg_t)receiverEndpointID << ENDPOINT_ID_SHIFT);
62 78
63 triggerCopy = localEndpoint->centralExchangePtr->triggerCounter +1; 79 triggerCopy = localEndpoint->centralExchange->triggerCounter +1;
64 msgDraft |= ((lossyCom__msg_t)triggerCopy << TRIGGER_SHIFT); 80 msgDraft |= ((lossyCom__msg_t)triggerCopy << TRIGGER_SHIFT);
65 81
66 //write msg to central exchange 82 //write msg to central exchange
67 localEndpoint->centralExchangePtr->outboxArray[localEndpoint->endpointID] = 83 localEndpoint->centralExchange->outboxArray[localEndpoint->endpointID] =
68 msgDraft; 84 msgDraft;
69 85
70 localEndpoint->centralExchangePtr->triggerCounter = triggerCopy; 86 localEndpoint->centralExchange->triggerCounter = triggerCopy;
71 } 87 }
72 88
73 void inline lossyCom__receiveMsg(lossyCom__endpoint_t* localEndpoint) 89 void inline lossyCom__receiveMsg(lossyCom__endpoint_t* localEndpoint)
74 { 90 {
75 uint16_t remoteTriggerCopy; 91 uint16_t currentTriggerCopy;
76 lossyCom__endpointID_t senderEndpointID; 92 lossyCom__endpointID_t senderEndpointID;
77 lossyCom__msg_t msgCopy; 93 lossyCom__msg_t msgCopy;
78 uint16_t msgTrigger; 94 uint16_t msgTrigger;
79 lossyCom__msgBody_t msgBody; 95 lossyCom__msgBody_t msgBody;
80 96
81 senderEndpointID = 0; 97 senderEndpointID = 0;
82 remoteTriggerCopy = localEndpoint->centralExchangePtr->triggerCounter; 98 currentTriggerCopy = localEndpoint->centralExchange->triggerCounter;
99
83 //new message arrived if trigger counter is higher than the last time read 100 //new message arrived if trigger counter is higher than the last time read
84 while(senderEndpointID < localEndpoint->centralExchangePtr->numEndpoints) 101 while(senderEndpointID < localEndpoint->centralExchange->numEndpoints)
85 { 102 {
86 if(senderEndpointID != localEndpoint->endpointID) 103 if(senderEndpointID != localEndpoint->endpointID)
87 { 104 {
88 msgCopy = localEndpoint->centralExchangePtr->outboxArray[senderEndpointID]; 105 msgCopy = localEndpoint->centralExchange->outboxArray[senderEndpointID];
89 msgTrigger = 0xFFFF & (msgCopy >> TRIGGER_SHIFT); 106 msgTrigger = 0xFFFF & (msgCopy >> TRIGGER_SHIFT);
107 // check if the message is new (msg trigger > archived trigger)
108 // and already valid (msgTrigger <= currentTriggerCopy)
90 if(msgTrigger > localEndpoint->localTriggerCopy && 109 if(msgTrigger > localEndpoint->localTriggerCopy &&
91 msgTrigger <= remoteTriggerCopy) 110 msgTrigger <= currentTriggerCopy)
92 { 111 {
112 //let the message handler parse the message
93 msgBody = 0xFFFFFFFF & msgCopy; 113 msgBody = 0xFFFFFFFF & msgCopy;
94 (*(localEndpoint->msgHandler))(senderEndpointID, msgBody); 114 (*(localEndpoint->msgHandler))(senderEndpointID,
115 msgBody,
116 localEndpoint->msgHandlerData);
95 } 117 }
96 } 118 }
97 senderEndpointID++; 119 senderEndpointID++;
98 } 120 }
99 //save last parsed msg 121 //save last parsed msg
100 localEndpoint->localTriggerCopy = remoteTriggerCopy; 122 localEndpoint->localTriggerCopy = currentTriggerCopy;
101 } 123 }