Mercurial > cgi-bin > hgwebdir.cgi > VMS > C_Libraries > BestEffortMessaging
view LossyCom.c @ 5:95a03e431480
corrected comments and variable renaming
| author | Merten Sach <msach@mailbox.tu-berlin.de> |
|---|---|
| date | Tue, 13 Mar 2012 19:07:49 +0100 |
| parents | 7ba5a3a6102d |
| children |
line source
1 /*
2 * For a detailed description see header file.
3 */
5 #include <string.h>
7 #include "LossyCom.h"
8 #include "VMS_Implementations/VMS_impl/vmalloc.h"
10 /*
11 * Initializes the central exchange structure and sets all trigger counter to 0
12 * Allocates memory to fit the number of endpoints.
13 * Returns NULL if an error occurs.
14 */
15 lossyCom__exchange_t* lossyCom__initialize(uint16_t numEndpoints)
16 {
17 lossyCom__exchange_t* centralExchange;
19 centralExchange = VMS_WL__malloc(sizeof(lossyCom__exchange_t));
20 if(centralExchange == NULL)
21 return NULL;
23 centralExchange->broadcastTriggerCounter = 0;
24 centralExchange->numEndpoints = numEndpoints;
25 centralExchange->outboxArray = VMS_WL__malloc(sizeof(lossyCom__msg_t)*numEndpoints);
26 if(centralExchange->outboxArray == NULL){
27 VMS_WL__free(centralExchange);
28 return NULL;
29 }
31 centralExchange->p2pTriggerCounters = VMS_WL__malloc(sizeof(uint16_t)*numEndpoints);
32 if(centralExchange->p2pTriggerCounters == NULL){
33 VMS_WL__free(centralExchange->outboxArray);
34 VMS_WL__free(centralExchange);
35 return NULL;
36 }
38 //reset all point 2 point trigger counter
39 memset((void*)centralExchange->p2pTriggerCounters, 0, sizeof(uint16_t)*numEndpoints);
41 return centralExchange;
42 }
44 /*
45 * Connects the local endpoint to the central exchange structure.
46 * This registers a message handler that handles all incoming messages.
47 * Also an endpointID is set this ID has to be between 0 and total number of
48 * endpoints-1 and unique.
49 */
50 void lossyCom__initialize_endpoint(lossyCom__endpoint_t* localEndpoint,
51 lossyCom__exchange_t* centralExchange,
52 lossyCom__endpointID_t endpointID,
53 lossyCom__msgHandler msgHandler,
54 void* msgHandlerData)
55 {
56 localEndpoint->lastReceivedBroadcastTrigger = 0;
57 localEndpoint->endpointID = endpointID;
58 localEndpoint->centralExchange = centralExchange;
59 localEndpoint->msgHandler = msgHandler;
60 localEndpoint->msgHandlerData = msgHandlerData;
61 }
63 /*
64 * Prepare a lossyCom__msg_t in the correct format that contains the trigger
65 * value, the receiver endpoint ID and the message body
66 */
67 inline lossyCom__msg_t prepareMsg(uint16_t triggerValue,
68 lossyCom__endpointID_t receiverEndpointID,
69 lossyCom__msgBody_t msgBody)
70 {
71 lossyCom__msg_t msgDraft;
73 msgDraft = (0 | msgBody);
74 msgDraft |= ((lossyCom__msg_t)receiverEndpointID << ENDPOINT_ID_SHIFT);
75 msgDraft |= ((lossyCom__msg_t)triggerValue << TRIGGER_SHIFT);
77 return msgDraft;
78 }
80 /*
81 * This broadcasts a message to all connected receivers
82 */
83 void lossyCom__broadcastMsg(lossyCom__endpoint_t* localEndpoint,
84 lossyCom__msgBody_t msgBody)
85 {
86 lossyCom__exchange_t* centralExchange = localEndpoint->centralExchange;
87 uint16_t increasedTrigger;
88 lossyCom__msg_t msg;
90 increasedTrigger = centralExchange->broadcastTriggerCounter +1;
92 //build message
93 msg = prepareMsg(increasedTrigger, BROADCAST_ID, msgBody);
95 //write msg to central exchange
96 centralExchange->outboxArray[localEndpoint->endpointID] = msg;
98 //update broadcast trigger counter
99 centralExchange->broadcastTriggerCounter = increasedTrigger;
100 }
102 /*
103 * This sends a message another endpoint. Again it is not guaranteed that the
104 * message is received. But in most cases it will.
105 */
106 void lossyCom__sendMsg(lossyCom__endpoint_t* localEndpoint,
107 lossyCom__endpointID_t receiverEndpointID,
108 lossyCom__msgBody_t msgBody)
109 {
110 lossyCom__exchange_t* centralExchange = localEndpoint->centralExchange;
111 lossyCom__msg_t msg;
112 uint16_t increasedTrigger;
114 increasedTrigger =
115 centralExchange->p2pTriggerCounters[receiverEndpointID] +1;
117 //build message
118 msg = prepareMsg(increasedTrigger, receiverEndpointID, msgBody);
120 //write msg to central exchange
121 centralExchange->outboxArray[localEndpoint->endpointID] = msg;
123 //write back increased trigger counter
124 centralExchange->p2pTriggerCounters[receiverEndpointID] = increasedTrigger;
125 }
127 int inline isUnreceivedMsg(uint16_t msgTrigger,
128 uint16_t lastReceivedTrigger,
129 uint16_t triggerSnapshot)
130 {
131 if(msgTrigger > lastReceivedTrigger ||
132 msgTrigger <= triggerSnapshot)
133 {
134 // check if the message is new (msg trigger > archived trigger)
135 // and already valid (msgTrigger <= currentTriggerCopy)
136 if((msgTrigger > lastReceivedTrigger &&
137 msgTrigger <= triggerSnapshot) ||
138 ((int64_t) triggerSnapshot- // check for triggerCounterOverflow
139 (int64_t)lastReceivedTrigger < -MAX_TRIGGER/2))
140 {
141 return TRUE;
142 }
143 }
144 return FALSE;
145 }
147 void lossyCom__receiveMsg(lossyCom__endpoint_t* localEndpoint)
148 {
149 uint16_t broadcastTriggerSnapshot, p2pTriggerSnapshot;
150 lossyCom__exchange_t* centralExchange;
151 lossyCom__endpointID_t senderEndpointID;
152 lossyCom__endpointID_t receiverEndpointID;
153 lossyCom__msg_t msgCopy;
154 uint16_t msgTrigger;
155 lossyCom__msgBody_t msgBody;
157 centralExchange = localEndpoint->centralExchange;
158 //save trigger counter to know find valid messages
159 broadcastTriggerSnapshot = centralExchange->broadcastTriggerCounter;
160 p2pTriggerSnapshot =
161 centralExchange->p2pTriggerCounters[localEndpoint->endpointID];
163 //new message arrived if trigger counter is higher than the last time read
164 if( broadcastTriggerSnapshot > localEndpoint->lastReceivedBroadcastTrigger ||
165 p2pTriggerSnapshot > localEndpoint->lastReceivedp2pTrigger)
166 {
167 senderEndpointID = 0;
168 //search outboxes for new messages
169 while(senderEndpointID < centralExchange->numEndpoints)
170 {
171 //ignore own outbox
172 if(senderEndpointID != localEndpoint->endpointID)
173 {
174 msgCopy = centralExchange->outboxArray[senderEndpointID];
175 msgTrigger = 0xFFFF & (msgCopy >> TRIGGER_SHIFT);
176 receiverEndpointID = 0xFFFF & (msgCopy >> ENDPOINT_ID_SHIFT);
178 if(receiverEndpointID == BROADCAST_ID){//receive broadcast message
179 if(isUnreceivedMsg(msgTrigger,
180 localEndpoint->lastReceivedBroadcastTrigger,
181 broadcastTriggerSnapshot))
182 {
183 //let the message handler parse the message
184 msgBody = 0xFFFFFFFF & msgCopy;
185 //only receive broadcast and p2p for own receiverID
187 (*(localEndpoint->msgHandler))(senderEndpointID,
188 msgBody,
189 localEndpoint->msgHandlerData);
190 }
191 }else{//point 2 point message
192 if(receiverEndpointID == localEndpoint->endpointID &&
193 isUnreceivedMsg(msgTrigger,
194 localEndpoint->lastReceivedp2pTrigger,
195 p2pTriggerSnapshot))
196 {
197 //let the message handler parse the message
198 msgBody = 0xFFFFFFFF & msgCopy;
199 //only receive broadcast and p2p for own receiverID
201 (*(localEndpoint->msgHandler))(senderEndpointID,
202 msgBody,
203 localEndpoint->msgHandlerData);
204 }
205 }
206 }
207 senderEndpointID++;
208 }//search outbox loop
209 }
210 //save last TriggerCounter of last parsed Msg
211 localEndpoint->lastReceivedBroadcastTrigger = broadcastTriggerSnapshot;
212 localEndpoint->lastReceivedp2pTrigger = p2pTriggerSnapshot;
213 }