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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
|
#include <stdlib.h>
#include <lua.h>
#include <lauxlib.h>
#include <lualib.h>
#define NNG_STATIC_LIB
#include <nng/nng.h>
#include <nng/transport/inproc/inproc.h>
#include <nng/transport/ipc/ipc.h>
#include <nng/transport/tcp/tcp.h>
#include <nng/transport/tls/tls.h>
#include <nng/transport/zerotier/zerotier.h>
#include <nng/protocol/pair1/pair.h>
#include <nng/protocol/bus0/bus.h>
#include <nng/protocol/pubsub0/pub.h>
#include <nng/protocol/pubsub0/sub.h>
#include <nng/protocol/pipeline0/pull.h>
#include <nng/protocol/pipeline0/push.h>
#include <nng/protocol/reqrep0/req.h>
#include <nng/protocol/reqrep0/rep.h>
#include <nng/protocol/survey0/respond.h>
#include <nng/protocol/survey0/survey.h>
#include <nng/supplemental/util/platform.h>
#include "lua-nng-common.h"
#include "lua-nng.h"
struct callback_info {
lua_State *L;
nng_mtx *lmutex; //mutex for the lua state
nng_mtx *cmutex;
nng_cv *cv;
int socketref; //lua refrence to this socket
nng_aio *aio;
int completed;
};
void push_callback(void *v){
/*printf("aio callback received!\n");*/
struct callback_info *ci = (struct callback_info*)v;
lua_State *L = ci->L;
/*printf("About to lock lua state in callback\n");*/
nng_mtx_lock(ci->lmutex);//lock the lua state
/*printf("Done locking lua state in callback\n");*/
int err = nng_aio_result(ci->aio);
if(err != 0){
/*printf("This callback was canceled or timed out: %d: %s\n", err, nng_strerror(err));*/
nng_mtx_unlock(ci->lmutex);
return;
}
lua_rawgeti(L,LUA_REGISTRYINDEX,ci->socketref);//push socket
luaL_unref(L,LUA_REGISTRYINDEX,ci->socketref);//free the reference
nng_msg *msg = nng_aio_get_msg(ci->aio);
size_t len = nng_msg_len(msg);
void *body = nng_msg_body(msg);
lua_pushlstring(L,(const char*)body, len);//push the message
nng_msg_free(msg);
ci->completed = 1;
lua_settable(L,1);
/*printf("About to unlock lua state in callback\n");*/
nng_mtx_unlock(ci->lmutex);
/*printf("About to lock condition mutex in callback\n");*/
nng_mtx_lock(ci->cmutex);
/*printf("About wake condition\n");*/
nng_cv_wake(ci->cv);
/*printf("Done wake condition\n");*/
nng_mtx_unlock(ci->cmutex);
/*printf("Done with callback\n");*/
}
//recv_any(socket1, socket2, ...) :: {socket = message}
int lnng_aio_recv(lua_State *L){
nng_mtx *luamtx, *callbackmtx;
int err = nng_mtx_alloc(&luamtx);
err |= nng_mtx_alloc(&callbackmtx);
/*err |= nng_mtx_alloc(&setupmtx);*/
if(err != 0){
/*printf("Something when wrong when allocating mutexes\n");*/
lua_pushboolean(L,0);
lua_pushstring(L,nng_strerror(err));
return 2;
}
int argv = lua_gettop(L);
/*printf("Receiving any on %d sockets\n",argv);*/
struct callback_info **cis = (struct callback_info**)malloc(sizeof(struct callback_info*) * argv);
nng_mtx_lock(luamtx);
/*printf("Locked lua state\n");*/
nng_cv *cv;
nng_cv_alloc(&cv, callbackmtx);
/*nng_mtx_lock(callbackmtx);//wait for one of the callbacks to happen*/
/*printf("Callback 1 happened\n");*/
for(int i = 0; i < argv; i++){
/*printf("\tSetting up async %d\n", i);*/
nng_socket *sock = tosocket(L,-1);
int sref = luaL_ref(L,LUA_REGISTRYINDEX);
/*printf("\tGot socket ref %d\n", sref);*/
cis[i] = (struct callback_info*)malloc(sizeof(struct callback_info));
struct callback_info *ci = cis[i];
/*printf("\tLooking at ci %p\n",ci);*/
ci->L = L;
ci->lmutex = luamtx;
ci->cmutex = callbackmtx;
ci->socketref = sref;
ci->completed = 0;
ci->cv = cv;
/*printf("\tAbout to alloc aio\n");*/
nng_aio_alloc(&(ci->aio), push_callback, ci);
/*printf("\tAllocated aio\n");*/
/*printf("\tEverything else set on callback info\n");*/
nng_recv_aio(*sock, ci->aio);
/*printf("\tSet up async receive %d\n",i);*/
}
lua_newtable(L);//table that will hold [socket] = message
/*printf("About to unlock lua state\n");*/
nng_mtx_unlock(luamtx);
/*printf("Unlocked lua state\n");*/
/*nng_mtx_lock(callbackmtx);//was unlocked by the callback, luamtx is locked at this point*/
int complete = 0;
nng_mtx_lock(callbackmtx);
while(complete == 0){
for(int i = 0; i < argv; i++){
struct callback_info *ci = cis[i];
if(ci->completed > 0){
/*printf("At least 1 completed! breaking!\n");*/
complete = 1;
goto found;
}
}
/*printf("About to wait\n");*/
nng_cv_wait(cv);
/*printf("Done waiting, complete is: %d\n", complete);*/
}
found:
/*printf("Callback 2 happened\n");*/
nng_mtx_unlock(callbackmtx);
/*printf("Callback done\n");*/
for(int i = 0; i < argv; i++){
struct callback_info *ci = cis[i];
/*printf("About to stop aio %d\n",ci->aio);*/
/*nng_aio_cancel(ci->aio);*/
nng_aio_stop(ci->aio);
nng_aio_free(ci->aio);
free(ci);
/*printf("Stopped aio %d\n",ci->aio);*/
}
free(cis);
/*printf("Freeing things\n");*/
nng_cv_free(cv);
nng_mtx_free(callbackmtx);
//nng_mtx_unlock(luamtx);//mutexes must not be locked when they are freed
nng_mtx_free(luamtx);
/*printf("Done freeing everything, returning...\n");*/
return 1;
}
static const struct luaL_Reg nng_aio_handler_m[] = {
{NULL, NULL}
};
static const struct luaL_Reg nng_aio_mutex_m[] = {
{NULL, NULL}
};
static const struct luaL_Reg nng_http_f[] = {
{"recv_any",lnng_aio_recv},
{NULL, NULL}
};
int luaopen_nng_aio(lua_State *L){
luaL_newmetatable(L,"nng.aio.struct");
luaL_newlib(L,nng_aio_handler_m);
lua_setfield(L,-2,"__index");
lua_pop(L,1);
luaL_newmetatable(L,"nng.aio.mutex");
luaL_newlib(L,nng_aio_mutex_m);
lua_setfield(L,-2,"__index");
lua_pop(L,1);
luaL_newlib(L,nng_http_f);
return 1;
}
|
