1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
#include "worker.h"
#include "nanomsg/nn.h"
#include "port.h"
#include <glib.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <signal.h>
const char *worker_exection_mode_str( worker_execution_mode_t mode )
{
switch( mode ) {
case WORKER_EXECUTION_DISABLED: return "disabled";
case WORKER_EXECUTION_DIRECT: return "direct";
default: return "<unknown>";
}
}
const char *worker_state_str( worker_state_t state )
{
switch( state ) {
case WORKER_STATE_STOPPED: return "stopped";
case WORKER_STATE_RUNNING: return "running";
default: return "<unknown>";
}
}
worker_execution_mode_t worker_execution_mode_from_str( const char *s )
{
if( strcasecmp( s, "disabled" ) == 0 ) {
return WORKER_EXECUTION_DISABLED;
} else if( strcasecmp( s, "direct" ) == 0 ) {
return WORKER_EXECUTION_DIRECT;
} else {
fprintf( stderr, "Warning: unknown worker execution mode '%s'!\n", s );
return WORKER_EXECUTION_DISABLED;
}
}
worker_state_t worker_state_from_str( const char *s )
{
if( strcasecmp( s, "stopped" ) == 0 ) {
return WORKER_STATE_STOPPED;
} else if( strcasecmp( s, "running" ) == 0 ) {
return WORKER_STATE_RUNNING;
} else {
fprintf( stderr, "Warning: unknown worker state '%s' (assuming 'stopped')!\n", s );
return WORKER_STATE_STOPPED;
}
}
typedef struct {
GPid pid;
gint out, err;
worker_t *worker;
pthread_t thread;
GMainLoop *main_loop;
} direct_glib_execution_worker_data_t;
static void watch_child( GPid pid, gint status, gpointer *data )
{
direct_glib_execution_worker_data_t *wed = (direct_glib_execution_worker_data_t *)data;
printf( "Worker child with PID %d terminated.\n", wed->pid );
g_spawn_close_pid( pid );
g_main_loop_quit( wed->main_loop );
}
static void *worker_func( void *thread_data )
{
direct_glib_execution_worker_data_t *wed = (direct_glib_execution_worker_data_t *)thread_data;
worker_t *worker = wed->worker;
gchar *argv[] = { worker->command, NULL };
gboolean ret = g_spawn_async_with_pipes( NULL,
argv, NULL, G_SPAWN_DO_NOT_REAP_CHILD, NULL,
NULL, &wed->pid, NULL, &wed->out, &wed->err, NULL );
if( !ret ) {
g_error( "Starting worker failed" );
return NULL;
}
printf( "Worker child with PID %d created: %s.\n",
wed->pid, worker->command );
GMainContext *context = g_main_context_default( );
wed->main_loop = g_main_loop_new( context, TRUE );
g_child_watch_add( wed->pid, (GChildWatchFunc)watch_child, (gpointer *)wed );
worker->state = WORKER_STATE_RUNNING;
printf( "worker %s: entering glib worker main loop..\n", worker->name );
g_main_loop_run( wed->main_loop );
printf( "worker %s: leaving glib worker main loop..\n", worker->name );
worker->state = WORKER_STATE_STOPPED;
return NULL;
}
int worker_init( worker_t *worker )
{
if( worker->state == WORKER_STATE_RUNNING ) {
return 0;
}
pthread_attr_t attr;
int res;
res = pthread_attr_init( &attr );
if( res != 0 ) {
return 1;
}
worker->execution_data = malloc( sizeof( direct_glib_execution_worker_data_t ) );
direct_glib_execution_worker_data_t *wed =
(direct_glib_execution_worker_data_t *)worker->execution_data;
wed->worker = worker;
res = pthread_create( &wed->thread, &attr, worker_func, (void *)wed );
if( res != 0 ) {
return 1;
}
pthread_detach( wed->thread );
return 0;
}
void worker_terminate( worker_t *worker )
{
direct_glib_execution_worker_data_t *wed = (direct_glib_execution_worker_data_t *)worker->execution_data;
#ifndef _WIN32
kill( wed->pid, SIGTERM );
#else
#error TODO kill on Windows
#endif
worker->state = WORKER_STATE_STOPPED;
free( worker->execution_data );
worker->execution_data = NULL;
}
int worker_free( worker_t *worker )
{
return 0;
}
|
