taskrambler / libtrcomm · Commits

Commit 4a64f7dd369ad36be71a66634ca927ae2fa2f55e

Authored by Georg Hopp
1 parent 52100363

completely drop edge triggerd behaviour as I don't see any real benefit in it an… 

…d it makes things unnessecary complicated.
Inline Side-by-side
Showing 6 changed files with 62 additions and 111 deletions
@@ -33,10 +33,8 @@ @@ -33,10 +33,8 @@
33 TR_CLASS(TR_CommManagerEpoll) { 33 TR_CLASS(TR_CommManagerEpoll) {
34 TR_EXTENDS(TR_CommManager); 34 TR_EXTENDS(TR_CommManager);
35 35
36 - int handle;  
37 - struct epoll_event * events;  
38 - TR_Queue read_ready;  
39 - TR_Queue write_ready; 36 + int handle;
  37 + uint32_t * events;
40 }; 38 };
41 TR_INSTANCE_INIT(TR_CommManagerEpoll); 39 TR_INSTANCE_INIT(TR_CommManagerEpoll);
42 TR_CLASSVARS_DECL(TR_CommManagerEpoll) { 40 TR_CLASSVARS_DECL(TR_CommManagerEpoll) {
@@ -31,21 +31,12 @@ TR_cepWriteBuffered(TR_CommEndPoint this) @@ -31,21 +31,12 @@ TR_cepWriteBuffered(TR_CommEndPoint this)
31 TR_RemoteData data; 31 TR_RemoteData data;
32 int send; 32 int send;
33 33
34 -// fprintf(stderr, "%s(%p): before get write data: %p / %zd messages\n",  
35 -// __func__, this, data, this->write_buffer->nmsg);  
36 data = TR_cepNextWriteData(this); 34 data = TR_cepNextWriteData(this);
37 -// fprintf(stderr, "%s(%p): get write data: %p / %zd messages\n",  
38 -// __func__, this, data, this->write_buffer->nmsg);  
39 -// fflush(stderr);  
40 -  
41 send = TR_socketSend(this->transport, data); 35 send = TR_socketSend(this->transport, data);
42 36
43 switch (send) { 37 switch (send) {
44 case FALSE: // EAGAIN 38 case FALSE: // EAGAIN
45 TR_queuePutFirst(this->write_buffer, data); 39 TR_queuePutFirst(this->write_buffer, data);
46 -// fprintf(stderr, "%s(%p): put first write data: %p / %zd messages\n",  
47 -// __func__, this, data, this->write_buffer->nmsg);  
48 -// fflush(stderr);  
49 break; 40 break;
50 41
51 case -1: // FAILURE 42 case -1: // FAILURE
@@ -47,12 +47,8 @@ commManagerEpollCtor(void * _this, va_list * params) @@ -47,12 +47,8 @@ commManagerEpollCtor(void * _this, va_list * params)
47 TR_CommManager cmgr = _this; 47 TR_CommManager cmgr = _this;
48 48
49 TR_PARENTCALL(TR_CommManagerEpoll, _this, TR_Class, ctor, params); 49 TR_PARENTCALL(TR_CommManagerEpoll, _this, TR_Class, ctor, params);
50 - this->handle = epoll_create(cmgr->n_endpoints);  
51 - this->read_ready = TR_new(TR_Queue);  
52 - this->write_ready = TR_new(TR_Queue);  
53 -  
54 - this->read_ready->free_msgs = 0;  
55 - this->write_ready->free_msgs = 0; 50 + this->handle = epoll_create(cmgr->n_endpoints);
  51 + this->events = TR_calloc(cmgr->n_endpoints, sizeof(uint32_t));
56 52
57 return 0; 53 return 0;
58 } 54 }
@@ -63,9 +59,7 @@ commManagerEpollDtor(void * _this) @@ -63,9 +59,7 @@ commManagerEpollDtor(void * _this)
63 { 59 {
64 TR_CommManagerEpoll this = _this; 60 TR_CommManagerEpoll this = _this;
65 61
66 - TR_delete(this->read_ready);  
67 - TR_delete(this->write_ready);  
68 - 62 + TR_MEM_FREE(this->events);
69 close(this->handle); 63 close(this->handle);
70 TR_PARENTCALL(TR_CommManagerEpoll, _this, TR_Class, dtor); 64 TR_PARENTCALL(TR_CommManagerEpoll, _this, TR_Class, dtor);
71 } 65 }
@@ -78,8 +72,9 @@ TR_commManagerEpollAddEndpoint(void * _this, TR_CommEndPoint endpoint) @@ -78,8 +72,9 @@ TR_commManagerEpollAddEndpoint(void * _this, TR_CommEndPoint endpoint)
78 int handle = endpoint->transport->handle; 72 int handle = endpoint->transport->handle;
79 struct epoll_event event; 73 struct epoll_event event;
80 74
81 - event.data.ptr = endpoint;  
82 - event.events = EPOLLIN | EPOLLOUT | EPOLLET; 75 + this->events[handle] = EPOLLIN;
  76 + event.data.ptr = endpoint;
  77 + event.events = this->events[handle];
83 78
84 epoll_ctl(this->handle, EPOLL_CTL_ADD, handle, &event); 79 epoll_ctl(this->handle, EPOLL_CTL_ADD, handle, &event);
85 } 80 }
@@ -90,11 +85,6 @@ TR_commManagerEpollSelect(void * _this, TR_Event event, int timeout) @@ -90,11 +85,6 @@ TR_commManagerEpollSelect(void * _this, TR_Event event, int timeout)
90 { 85 {
91 TR_CommManagerEpoll this = _this; 86 TR_CommManagerEpoll this = _this;
92 int i, nevents; 87 int i, nevents;
93 - TR_Queue node;  
94 -  
95 - if (0 != (this->read_ready->nmsg & this->write_ready->nmsg)) {  
96 - timeout = 0;  
97 - }  
98 88
99 nevents = epoll_wait(this->handle, events, MAXEVENTS, timeout); 89 nevents = epoll_wait(this->handle, events, MAXEVENTS, timeout);
100 90
@@ -110,40 +100,16 @@ TR_commManagerEpollSelect(void * _this, TR_Event event, int timeout) @@ -110,40 +100,16 @@ TR_commManagerEpollSelect(void * _this, TR_Event event, int timeout)
110 TR_CET_EVENT_ACC_READY, 100 TR_CET_EVENT_ACC_READY,
111 NULL)); 101 NULL));
112 } else { 102 } else {
113 - if (! ((TR_EventSubject)endpoint)->fin) {  
114 - TR_queuePut(this->read_ready, endpoint);  
115 - } 103 + TR_eventHandlerIssueEvent((TR_EventHandler)this,
  104 + TR_eventSubjectEmit(
  105 + (TR_EventSubject)endpoint,
  106 + TR_CEP_EVENT_READ_READY,
  107 + NULL));
116 } 108 }
117 } 109 }
118 110
119 if ((events[i].events & EPOLLOUT) == EPOLLOUT) { 111 if ((events[i].events & EPOLLOUT) == EPOLLOUT) {
120 - if (TR_cepHasPendingData(endpoint) &&  
121 - ! ((TR_EventSubject)endpoint)->fin) {  
122 - TR_queuePut(this->write_ready, endpoint);  
123 - }  
124 - }  
125 - }  
126 -  
127 - /* now issue reads and write events */  
128 - for (node=this->read_ready->first; node; node=node->next) {  
129 - TR_CommEndPoint endpoint = (TR_CommEndPoint)node->msg;  
130 -  
131 - if (! TR_socketFinRd(endpoint->transport)) {  
132 - TR_eventHandlerIssueEvent(  
133 - (TR_EventHandler)this,  
134 - TR_eventSubjectEmit(  
135 - (TR_EventSubject)endpoint,  
136 - TR_CEP_EVENT_READ_READY,  
137 - NULL));  
138 - }  
139 - }  
140 -  
141 - for (node=this->write_ready->first; node; node=node->next) {  
142 - TR_CommEndPoint endpoint = (TR_CommEndPoint)node->msg;  
143 -  
144 - if (! TR_socketFinWr(endpoint->transport)) {  
145 - TR_eventHandlerIssueEvent(  
146 - (TR_EventHandler)this, 112 + TR_eventHandlerIssueEvent((TR_EventHandler)this,
147 TR_eventSubjectEmit( 113 TR_eventSubjectEmit(
148 (TR_EventSubject)endpoint, 114 (TR_EventSubject)endpoint,
149 TR_CEP_EVENT_WRITE_READY, 115 TR_CEP_EVENT_WRITE_READY,
@@ -154,80 +120,84 @@ TR_commManagerEpollSelect(void * _this, TR_Event event, int timeout) @@ -154,80 +120,84 @@ TR_commManagerEpollSelect(void * _this, TR_Event event, int timeout)
154 120
155 static 121 static
156 void 122 void
157 -TR_commManagerEpollRemoveWrite(void * _this, TR_Event event) 123 +TR_commManagerEpollEnableWrite(void * _this, TR_Event event)
158 { 124 {
159 TR_CommManagerEpoll this = _this; 125 TR_CommManagerEpoll this = _this;
160 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject; 126 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject;
161 127
162 - TR_queueDelete(this->write_ready, endpoint); 128 + if (! TR_socketFinWr(endpoint->transport)) {
  129 + int handle = endpoint->transport->handle;
  130 + struct epoll_event _event;
  131 +
  132 + this->events[handle] |= EPOLLOUT;
  133 + _event.data.ptr = endpoint;
  134 + _event.events = this->events[handle];
  135 +
  136 + epoll_ctl(this->handle, EPOLL_CTL_MOD, handle, &_event);
  137 + }
163 } 138 }
164 139
165 static 140 static
166 void 141 void
167 -TR_commManagerEpollRemoveRead(void * _this, TR_Event event) 142 +TR_commManagerEpollDisableWrite(void * _this, TR_Event event)
168 { 143 {
169 TR_CommManagerEpoll this = _this; 144 TR_CommManagerEpoll this = _this;
170 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject; 145 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject;
  146 + int handle = endpoint->transport->handle;
  147 + struct epoll_event _event;
  148 +
  149 + this->events[handle] &= ~EPOLLOUT;
  150 + _event.data.ptr = endpoint;
  151 + _event.events = this->events[handle];
171 152
172 - TR_queueDelete(this->read_ready, endpoint); 153 + epoll_ctl(this->handle, EPOLL_CTL_MOD, handle, &_event);
173 } 154 }
174 155
175 static 156 static
176 void 157 void
177 -TR_commManagerEpollClose(void * _this, TR_Event event) 158 +TR_commManagerEpollEnableRead(void * _this, TR_Event event)
178 { 159 {
179 TR_CommManagerEpoll this = _this; 160 TR_CommManagerEpoll this = _this;
180 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject; 161 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject;
181 162
182 - TR_queueDelete(this->read_ready, endpoint);  
183 - TR_queueDelete(this->write_ready, endpoint); 163 + if (! TR_socketFinRd(endpoint->transport)) {
  164 + int handle = endpoint->transport->handle;
  165 + struct epoll_event _event;
184 166
185 - epoll_ctl(this->handle, EPOLL_CTL_DEL, endpoint->transport->handle, NULL); 167 + this->events[handle] |= EPOLLIN;
  168 + _event.data.ptr = endpoint;
  169 + _event.events = this->events[handle];
  170 +
  171 + epoll_ctl(this->handle, EPOLL_CTL_MOD, handle, &_event);
  172 + }
186 } 173 }
187 174
188 static 175 static
189 void 176 void
190 -TR_commManagerEpollShutRead(void * _this, TR_Event event) 177 +TR_commManagerEpollDisableRead(void * _this, TR_Event event)
191 { 178 {
192 TR_CommManagerEpoll this = _this; 179 TR_CommManagerEpoll this = _this;
193 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject; 180 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject;
  181 + int handle = endpoint->transport->handle;
194 struct epoll_event _event; 182 struct epoll_event _event;
195 183
196 - TR_queueDelete(this->read_ready, endpoint);  
197 - 184 + this->events[handle] &= ~EPOLLIN;
198 _event.data.ptr = endpoint; 185 _event.data.ptr = endpoint;
199 - _event.events = EPOLLOUT | EPOLLET; 186 + _event.events = this->events[handle];
200 187
201 - epoll_ctl(  
202 - this->handle,  
203 - EPOLL_CTL_MOD,  
204 - endpoint->transport->handle,  
205 - &_event); 188 + epoll_ctl(this->handle, EPOLL_CTL_MOD, handle, &_event);
206 } 189 }
207 190
208 static 191 static
209 void 192 void
210 -TR_commManagerEpollShutWrite(void * _this, TR_Event event) 193 +TR_commManagerEpollClose(void * _this, TR_Event event)
211 { 194 {
212 TR_CommManagerEpoll this = _this; 195 TR_CommManagerEpoll this = _this;
213 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject; 196 TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject;
214 - struct epoll_event _event;  
215 -  
216 - TR_queueDelete(this->write_ready, endpoint);  
217 197
218 - _event.data.ptr = endpoint;  
219 - _event.events = EPOLLIN | EPOLLET;  
220 -  
221 - epoll_ctl(  
222 - this->handle,  
223 - EPOLL_CTL_MOD,  
224 - endpoint->transport->handle,  
225 - &_event); 198 + epoll_ctl(this->handle, EPOLL_CTL_DEL, endpoint->transport->handle, NULL);
226 } 199 }
227 200
228 -  
229 -static void TR_commManagerEpollNoop(void * _this, TR_Event event) {}  
230 -  
231 static 201 static
232 void 202 void
233 TR_commManagerEpollCvInit(TR_class_ptr cls) { 203 TR_commManagerEpollCvInit(TR_class_ptr cls) {
@@ -237,14 +207,14 @@ TR_commManagerEpollCvInit(TR_class_ptr cls) { @@ -237,14 +207,14 @@ TR_commManagerEpollCvInit(TR_class_ptr cls) {
237 TR_INIT_IFACE(TR_Class, commManagerEpollCtor, commManagerEpollDtor, NULL); 207 TR_INIT_IFACE(TR_Class, commManagerEpollCtor, commManagerEpollDtor, NULL);
238 TR_INIT_IFACE( 208 TR_INIT_IFACE(
239 TR_CommManager, 209 TR_CommManager,
240 - TR_commManagerEpollAddEndpoint, // TR_CON_EVENT_NEW_CON  
241 - TR_commManagerEpollSelect, // TR_DISPATCHER_EVENT_DATA_WAIT  
242 - TR_commManagerEpollRemoveWrite, // TR_CEP_EVENT_PENDING_DATA => WRITE_BLOCK  
243 - TR_commManagerEpollNoop, // TR_CEP_EVENT_END_DATA  
244 - TR_commManagerEpollRemoveRead, // TR_CEP_EVENT_READ_BLOCK  
245 - TR_commManagerEpollClose, // TR_CEP_EVENT_CLOSE  
246 - TR_commManagerEpollShutWrite, // TR_CEP_EVENT_SHUT_READ  
247 - TR_commManagerEpollShutRead); // TR_CEP_EVENT_SHUT_WRITE 210 + TR_commManagerEpollAddEndpoint, // TR_CON_EVENT_NEW_CON
  211 + TR_commManagerEpollSelect, // TR_DISPATCHER_EVENT_DATA_WAIT
  212 + TR_commManagerEpollEnableWrite, // TR_CEP_EVENT_PENDING_DATA => WRITE_BLOCK
  213 + TR_commManagerEpollDisableWrite, // TR_CEP_EVENT_END_DATA
  214 + TR_commManagerEpollEnableRead, // TR_CEP_EVENT_READ_BLOCK
  215 + TR_commManagerEpollClose, // TR_CEP_EVENT_CLOSE
  216 + TR_commManagerEpollDisableWrite, // TR_CEP_EVENT_SHUT_READ
  217 + TR_commManagerEpollEnableRead); // TR_CEP_EVENT_SHUT_WRITE
248 TR_CREATE_CLASS( 218 TR_CREATE_CLASS(
249 TR_CommManagerEpoll, 219 TR_CommManagerEpoll,
250 TR_CommManager, 220 TR_CommManager,
@@ -123,11 +123,6 @@ connectionCompose(void * _this, TR_ProtoMessage message) @@ -123,11 +123,6 @@ connectionCompose(void * _this, TR_ProtoMessage message)
123 } 123 }
124 124
125 TR_queuePut(((TR_CommEndPoint)_this)->write_buffer, data); 125 TR_queuePut(((TR_CommEndPoint)_this)->write_buffer, data);
126 -// fprintf(stderr, "%s(%p): put write data: %p / %zd messages\n",  
127 -// __func__, (TR_CommEndPoint)_this, data,  
128 -// ((TR_CommEndPoint)_this)->write_buffer->nmsg);  
129 -// fflush(stderr);  
130 -  
131 return TRUE; 126 return TRUE;
132 } 127 }
133 128
@@ -110,10 +110,6 @@ datagramServiceCompose(void * _this, TR_ProtoMessage message) @@ -110,10 +110,6 @@ datagramServiceCompose(void * _this, TR_ProtoMessage message)
110 } 110 }
111 111
112 TR_queuePut(((TR_CommEndPoint)_this)->write_buffer, data); 112 TR_queuePut(((TR_CommEndPoint)_this)->write_buffer, data);
113 -// fprintf(stderr, "%s(%p): put write data: %p / %zd messages\n",  
114 -// __func__, (TR_CommEndPoint)_this, data,  
115 -// ((TR_CommEndPoint)_this)->write_buffer->nmsg);  
116 -// fflush(stderr);  
117 return TRUE; 113 return TRUE;
118 } 114 }
119 115
@@ -86,10 +86,11 @@ static @@ -86,10 +86,11 @@ static
86 TR_EventDone 86 TR_EventDone
87 ioHandlerWrite(void * _this, TR_Event event) 87 ioHandlerWrite(void * _this, TR_Event event)
88 { 88 {
89 - TR_Event revent = NULL,  
90 - close_event = NULL; 89 + TR_Event revent = NULL,
  90 + close_event = NULL;
  91 + TR_CommEndPoint endpoint = (TR_CommEndPoint)event->subject;
91 92
92 - switch (TR_cepWriteBuffered((TR_CommEndPoint)event->subject)) { 93 + switch (TR_cepWriteBuffered(endpoint)) {
93 case FALSE: // EAGAIN 94 case FALSE: // EAGAIN
94 revent = TR_eventSubjectEmit( 95 revent = TR_eventSubjectEmit(
95 event->subject, 96 event->subject,
Please register or login to post a comment