taskrambler / libtrevent · Commits

Commit 9e0658637ee7c380e2c2e35cdbcf66e92ca03f4b

Authored by Georg Hopp
1 parent 48a57a14

Revert "Revert "more thread syncs."" 

This reverts commit 48a57a14.
Inline Side-by-side
Showing 13 changed files with 102 additions and 32 deletions
@@ -52,7 +52,8 @@ TR_CLASS(TR_EventDispatcher) { @@ -52,7 +52,8 @@ TR_CLASS(TR_EventDispatcher) {
52 TR_Queue events; 52 TR_Queue events;
53 pthread_mutex_t events_lock; 53 pthread_mutex_t events_lock;
54 pthread_cond_t events_cond; 54 pthread_cond_t events_cond;
55 - pthread_t events_wait; 55 + pthread_t events_poll;
  56 + size_t events_handling;
56 57
57 TR_Hash handler; 58 TR_Hash handler;
58 TR_EventHandler default_handler; 59 TR_EventHandler default_handler;
@@ -24,6 +24,7 @@ @@ -24,6 +24,7 @@
24 #define __TR_EVENT_HANDLER_H__ 24 #define __TR_EVENT_HANDLER_H__
25 25
26 #include <sys/types.h> 26 #include <sys/types.h>
  27 +#include <pthread.h>
27 28
28 #include "trbase.h" 29 #include "trbase.h"
29 #include "trdata.h" 30 #include "trdata.h"
@@ -68,6 +69,12 @@ void TR__eventHandlerClassCleanup(TR_class_ptr); @@ -68,6 +69,12 @@ void TR__eventHandlerClassCleanup(TR_class_ptr);
68 sizeof(TR_EventMethod_fptr))); \ 69 sizeof(TR_EventMethod_fptr))); \
69 } while(0) 70 } while(0)
70 71
  72 +#define TR_INIT_HANDLER(cname) \
  73 + TR_INSTANCE(TR_Tree, cname##EventMethodsTree, NULL, PTHREAD_MUTEX_INITIALIZER); \
  74 + TR_INSTANCE(TR_Hash, cname##EventMethods, &(_##cname##EventMethodsTree.data), 0)
  75 +
  76 +#define TR_HANDLER_CVARS(cname) &(_##cname##EventMethods.data)
  77 +
71 #endif // __TR_EVENT_HANDLER_H__ 78 #endif // __TR_EVENT_HANDLER_H__
72 79
73 // vim: set ts=4 sw=4: 80 // vim: set ts=4 sw=4:
@@ -25,12 +25,14 @@ @@ -25,12 +25,14 @@
25 25
26 #include <sys/types.h> 26 #include <sys/types.h>
27 #include <stdint.h> 27 #include <stdint.h>
  28 +#include <pthread.h>
28 29
29 #include "trbase.h" 30 #include "trbase.h"
30 31
31 TR_CLASS(TR_EventSubject) { 32 TR_CLASS(TR_EventSubject) {
32 - int fin;  
33 - size_t emitted; 33 + int fin;
  34 + size_t emitted;
  35 + pthread_mutex_t lock;
34 }; 36 };
35 TR_INSTANCE_INIT(TR_EventSubject); 37 TR_INSTANCE_INIT(TR_EventSubject);
36 TR_CLASSVARS_DECL(TR_EventSubject) { 38 TR_CLASSVARS_DECL(TR_EventSubject) {
@@ -57,6 +59,16 @@ TR_CLASSVARS_DECL(TR_EventSubject) { @@ -57,6 +59,16 @@ TR_CLASSVARS_DECL(TR_EventSubject) {
57 intptr_t TR__eventSubjectId(TR_class_ptr, size_t); 59 intptr_t TR__eventSubjectId(TR_class_ptr, size_t);
58 TR_Event TR_eventSubjectEmit(TR_EventSubject, int, void *); 60 TR_Event TR_eventSubjectEmit(TR_EventSubject, int, void *);
59 61
  62 +static
  63 +inline
  64 +void
  65 +TR_eventSubjectAbsorb(TR_EventSubject this)
  66 +{
  67 + pthread_mutex_lock(&this->lock);
  68 + this->emitted = this->emitted == 0 ? 0 : this->emitted - 1;
  69 + pthread_mutex_unlock(&this->lock);
  70 +}
  71 +
60 #define TR_eventSubjectFinalize(es) ((es)->fin = TRUE) 72 #define TR_eventSubjectFinalize(es) ((es)->fin = TRUE)
61 73
62 #endif // __TR_EVENT_SUBJECT_H__ 74 #endif // __TR_EVENT_SUBJECT_H__
@@ -29,8 +29,9 @@ @@ -29,8 +29,9 @@
29 #include "event_dispatcher.h" 29 #include "event_dispatcher.h"
30 30
31 TR_CLASS(TR_EventThread) { 31 TR_CLASS(TR_EventThread) {
32 - TR_EventDispatcher dispatcher;  
33 - pthread_t handle; 32 + TR_EventDispatcher dispatcher;
  33 + pthread_t handle;
  34 + char * name;
34 }; 35 };
35 TR_INSTANCE_INIT(TR_EventThread); 36 TR_INSTANCE_INIT(TR_EventThread);
36 TR_CLASSVARS_DECL(TR_EventThread) {}; 37 TR_CLASSVARS_DECL(TR_EventThread) {};
@@ -41,13 +41,13 @@ eventCtor(void * _this, va_list * params) @@ -41,13 +41,13 @@ eventCtor(void * _this, va_list * params)
41 return 0; 41 return 0;
42 } 42 }
43 43
44 -static void eventDtor(void * _this) 44 +static
  45 +void
  46 +eventDtor(void * _this)
45 { 47 {
46 TR_Event this = _this; 48 TR_Event this = _this;
47 49
48 - this->subject->emitted =  
49 - this->subject->emitted == 0  
50 - ? 0 : this->subject->emitted - 1; 50 + TR_eventSubjectAbsorb(this->subject);
51 51
52 if (0 == this->subject->emitted && this->subject->fin) { 52 if (0 == this->subject->emitted && this->subject->fin) {
53 TR_delete(this->subject); 53 TR_delete(this->subject);
@@ -57,6 +57,7 @@ init_signals(void) @@ -57,6 +57,7 @@ init_signals(void)
57 signal(SIGABRT, terminate); 57 signal(SIGABRT, terminate);
58 signal(SIGALRM, SIG_IGN); 58 signal(SIGALRM, SIG_IGN);
59 signal(SIGURG, SIG_IGN); 59 signal(SIGURG, SIG_IGN);
  60 + signal(SIGUSR1, SIG_IGN);
60 61
61 signal(SIGPIPE, SIG_IGN); 62 signal(SIGPIPE, SIG_IGN);
62 } 63 }
@@ -20,6 +20,8 @@ @@ -20,6 +20,8 @@
20 * along with this program. If not, see <http://www.gnu.org/licenses/>. 20 * along with this program. If not, see <http://www.gnu.org/licenses/>.
21 */ 21 */
22 22
  23 +#define _GNU_SOURCE
  24 +
23 #include "trbase.h" 25 #include "trbase.h"
24 #include "trdata.h" 26 #include "trdata.h"
25 #include "trhash.h" 27 #include "trhash.h"
@@ -49,28 +51,39 @@ TR_eventDispatcherStart(TR_EventDispatcher this) @@ -49,28 +51,39 @@ TR_eventDispatcherStart(TR_EventDispatcher this)
49 (TR_EventSubject)this, 51 (TR_EventSubject)this,
50 TR_DISPATCHER_EVENT_HEARTBEAT, 52 TR_DISPATCHER_EVENT_HEARTBEAT,
51 NULL); 53 NULL);
52 - } else if (TR_queueEmpty(this->events)) {  
53 - if (! this->events_wait) { 54 + } else {
  55 + event = TR_queueGet(this->events);
  56 +
  57 + if (! (event || this->events_poll || this->events_handling)) {
54 int evtid = TR_EVD_CLIENT == this->mode 58 int evtid = TR_EVD_CLIENT == this->mode
55 ? TR_DISPATCHER_EVENT_USER_WAIT 59 ? TR_DISPATCHER_EVENT_USER_WAIT
56 : TR_DISPATCHER_EVENT_DATA_WAIT; 60 : TR_DISPATCHER_EVENT_DATA_WAIT;
57 61
58 - this->events_wait = pthread_self(); 62 + this->events_poll = pthread_self();
59 event = TR_eventSubjectEmit((TR_EventSubject)this, evtid, NULL); 63 event = TR_eventSubjectEmit((TR_EventSubject)this, evtid, NULL);
60 - } else {  
61 - pthread_cond_wait(&(this->events_cond), &(this->events_lock));  
62 - event = NULL;  
63 } 64 }
64 - } else {  
65 - event = TR_queueGet(this->events);  
66 } 65 }
67 66
68 - pthread_mutex_unlock(&(this->events_lock));  
69 -  
70 if (! event) { 67 if (! event) {
  68 + char buffer[17];
  69 +
  70 + pthread_getname_np(pthread_self(), buffer, 17);
  71 + TR_loggerLog(TR_logger, TR_LOGGER_DEBUG,
  72 + "[%s] - enter cond wait",
  73 + buffer);
  74 + pthread_cond_wait(&(this->events_cond), &(this->events_lock));
  75 + TR_loggerLog(TR_logger, TR_LOGGER_DEBUG,
  76 + "[%s] - leave cond wait",
  77 + buffer);
  78 + event = NULL;
  79 + pthread_mutex_unlock(&(this->events_lock));
71 continue; 80 continue;
  81 + } else {
  82 + this->events_handling++;
72 } 83 }
73 84
  85 + pthread_mutex_unlock(&(this->events_lock));
  86 +
74 handler_queue_hv = TR_hashGetByVal( 87 handler_queue_hv = TR_hashGetByVal(
75 this->handler, 88 this->handler,
76 TR_sdbm( 89 TR_sdbm(
@@ -104,9 +117,12 @@ TR_eventDispatcherStart(TR_EventDispatcher this) @@ -104,9 +117,12 @@ TR_eventDispatcherStart(TR_EventDispatcher this)
104 TR_delete(event); 117 TR_delete(event);
105 } 118 }
106 119
107 - if (pthread_equal(this->events_wait, pthread_self())) {  
108 - this->events_wait = FALSE; 120 + pthread_mutex_lock(&(this->events_lock));
  121 + this->events_handling--;
  122 + if (pthread_equal(this->events_poll, pthread_self())) {
  123 + this->events_poll = FALSE;
109 } 124 }
  125 + pthread_mutex_unlock(&(this->events_lock));
110 } 126 }
111 } 127 }
112 128
@@ -20,6 +20,8 @@ @@ -20,6 +20,8 @@
20 * along with this program. If not, see <http://www.gnu.org/licenses/>. 20 * along with this program. If not, see <http://www.gnu.org/licenses/>.
21 */ 21 */
22 22
  23 +#define _GNU_SOURCE
  24 +
23 #include <stdio.h> 25 #include <stdio.h>
24 26
25 #include "trbase.h" 27 #include "trbase.h"
@@ -32,7 +34,7 @@ @@ -32,7 +34,7 @@
32 TR_EventDone 34 TR_EventDone
33 TR_eventHandlerHandleEvent(TR_EventHandler this, TR_Event event) 35 TR_eventHandlerHandleEvent(TR_EventHandler this, TR_Event event)
34 { 36 {
35 - TR_EventDone retval; 37 + char buffer[17];
36 TR_EventMethod_fptr event_func = NULL; 38 TR_EventMethod_fptr event_func = NULL;
37 TR_HashValue handle_func_hv = TR_hashGetByVal( 39 TR_HashValue handle_func_hv = TR_hashGetByVal(
38 TR_CLASSVARS(TR_EventHandler, TR_GET_CLASS(this))->event_methods, 40 TR_CLASSVARS(TR_EventHandler, TR_GET_CLASS(this))->event_methods,
@@ -44,17 +46,17 @@ TR_eventHandlerHandleEvent(TR_EventHandler this, TR_Event event) @@ -44,17 +46,17 @@ TR_eventHandlerHandleEvent(TR_EventHandler this, TR_Event event)
44 46
45 event_func = *(TR_EventMethod_fptr *)handle_func_hv->value; 47 event_func = *(TR_EventMethod_fptr *)handle_func_hv->value;
46 48
47 - retval = event_func(this, event); 49 + pthread_getname_np(pthread_self(), buffer, 17);
48 50
49 TR_loggerLog(TR_logger, TR_LOGGER_DEBUG, 51 TR_loggerLog(TR_logger, TR_LOGGER_DEBUG,
50 - "[%ld] - HANDLE(%zd): %s event on %p with no. %d",  
51 - pthread_self(), 52 + "[%s] - HANDLE(%zd): %s event on %p with no. %d",
  53 + buffer,
52 event->subject->emitted, 54 event->subject->emitted,
53 TR_getEventString(event), 55 TR_getEventString(event),
54 event->subject, 56 event->subject,
55 event->serial); 57 event->serial);
56 58
57 - return retval; 59 + return event_func(this, event);
58 } 60 }
59 61
60 // vim: set ts=4 sw=4: 62 // vim: set ts=4 sw=4:
@@ -20,6 +20,8 @@ @@ -20,6 +20,8 @@
20 * along with this program. If not, see <http://www.gnu.org/licenses/>. 20 * along with this program. If not, see <http://www.gnu.org/licenses/>.
21 */ 21 */
22 22
  23 +#define _GNU_SOURCE
  24 +
23 #include <stdio.h> 25 #include <stdio.h>
24 #include <pthread.h> 26 #include <pthread.h>
25 27
@@ -32,12 +34,15 @@ int @@ -32,12 +34,15 @@ int
32 TR_eventHandlerIssueEvent(TR_EventHandler this, TR_Event event) 34 TR_eventHandlerIssueEvent(TR_EventHandler this, TR_Event event)
33 { 35 {
34 if (event) { 36 if (event) {
35 - int i; 37 + int i;
  38 + char buffer[17];
  39 +
  40 + pthread_getname_np(pthread_self(), buffer, 17);
36 41
37 for (i=0; i<this->ndispatcher; i++) { 42 for (i=0; i<this->ndispatcher; i++) {
38 TR_loggerLog(TR_logger, TR_LOGGER_DEBUG, 43 TR_loggerLog(TR_logger, TR_LOGGER_DEBUG,
39 - "[%ld] - ISSUE(%zd): %s event on %p with no. %d",  
40 - pthread_self(), 44 + "[%s] - ISSUE(%zd): %s event on %p with no. %d",
  45 + buffer,
41 event->subject->emitted, 46 event->subject->emitted,
42 TR_getEventString(event), 47 TR_getEventString(event),
43 event->subject, 48 event->subject,
@@ -27,8 +27,21 @@ @@ -27,8 +27,21 @@
27 #include "tr/logger.h" 27 #include "tr/logger.h"
28 #include "trbase.h" 28 #include "trbase.h"
29 29
30 -static int eventSubjectCtor(void * _this, va_list * params) { return 0; }  
31 -static void eventSubjectDtor(void * _this) {} 30 +static
  31 +int
  32 +eventSubjectCtor(void * _this, va_list * params)
  33 +{
  34 + pthread_mutex_init(&((TR_EventSubject)_this)->lock, NULL);
  35 +
  36 + return 0;
  37 +}
  38 +
  39 +static
  40 +void
  41 +eventSubjectDtor(void * _this)
  42 +{
  43 + pthread_mutex_destroy(&((TR_EventSubject)_this)->lock);
  44 +}
32 45
33 TR_INIT_IFACE(TR_Class, eventSubjectCtor, eventSubjectDtor, NULL); 46 TR_INIT_IFACE(TR_Class, eventSubjectCtor, eventSubjectDtor, NULL);
34 TR_CREATE_CLASS(TR_EventSubject, NULL, NULL, TR_IF(TR_Class)) = { 47 TR_CREATE_CLASS(TR_EventSubject, NULL, NULL, TR_IF(TR_Class)) = {
@@ -22,6 +22,7 @@ @@ -22,6 +22,7 @@
22 22
23 #include <stdio.h> 23 #include <stdio.h>
24 #include <stdint.h> 24 #include <stdint.h>
  25 +#include <pthread.h>
25 26
26 #include "tr/event.h" 27 #include "tr/event.h"
27 #include "tr/event_subject.h" 28 #include "tr/event_subject.h"
@@ -37,7 +38,9 @@ TR_eventSubjectEmit(TR_EventSubject this, int idx, void * data) @@ -37,7 +38,9 @@ TR_eventSubjectEmit(TR_EventSubject this, int idx, void * data)
37 if (id && ! this->fin) { 38 if (id && ! this->fin) {
38 event = TR_new(TR_Event, id, this); 39 event = TR_new(TR_Event, id, this);
39 TR_eventSetData(event, data); 40 TR_eventSetData(event, data);
  41 + pthread_mutex_lock(&this->lock);
40 this->emitted++; 42 this->emitted++;
  43 + pthread_mutex_unlock(&this->lock);
41 } 44 }
42 45
43 return event; 46 return event;
@@ -31,9 +31,13 @@ static @@ -31,9 +31,13 @@ static
31 int 31 int
32 eventThreadCtor(void * _this, va_list * params) 32 eventThreadCtor(void * _this, va_list * params)
33 { 33 {
34 - TR_EventThread this = _this; 34 + TR_EventThread this = _this;
  35 + char * name;
35 36
36 this->dispatcher = va_arg(*params, TR_EventDispatcher); 37 this->dispatcher = va_arg(*params, TR_EventDispatcher);
  38 + name = va_arg(*params, char *);
  39 +
  40 + if (name) this->name = TR_strdup(name);
37 41
38 return 0; 42 return 0;
39 } 43 }
@@ -48,6 +52,8 @@ eventThreadDtor(void * _this) @@ -48,6 +52,8 @@ eventThreadDtor(void * _this)
48 pthread_cancel(this->handle); 52 pthread_cancel(this->handle);
49 pthread_join(this->handle, NULL); 53 pthread_join(this->handle, NULL);
50 } 54 }
  55 +
  56 + TR_MEM_FREE(this->name);
51 } 57 }
52 58
53 TR_INIT_IFACE(TR_Class, eventThreadCtor, eventThreadDtor, NULL); 59 TR_INIT_IFACE(TR_Class, eventThreadCtor, eventThreadDtor, NULL);
@@ -20,6 +20,8 @@ @@ -20,6 +20,8 @@
20 * along with this program. If not, see <http://www.gnu.org/licenses/>. 20 * along with this program. If not, see <http://www.gnu.org/licenses/>.
21 */ 21 */
22 22
  23 +#define _GNU_SOURCE
  24 +
23 #include <pthread.h> 25 #include <pthread.h>
24 26
25 #include "trbase.h" 27 #include "trbase.h"
@@ -32,6 +34,7 @@ void * @@ -32,6 +34,7 @@ void *
32 TR_eventStreadRun(void * message) 34 TR_eventStreadRun(void * message)
33 { 35 {
34 TR_EventThread this = message; 36 TR_EventThread this = message;
  37 + if (this->name) pthread_setname_np(pthread_self(), this->name);
35 TR_eventDispatcherStart(this->dispatcher); 38 TR_eventDispatcherStart(this->dispatcher);
36 return NULL; 39 return NULL;
37 } 40 }
Please register or login to post a comment