pipe.c 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477
  1. /*
  2. ** The pipe algo is somewhat similar to the luaL_Buffer algo, except that it
  3. ** uses table to store the LUAL_BUFFERSIZE byte array chunks instead of the
  4. ** stack. Writing is always to the last UD in the table and overflow pushes a
  5. ** new UD to the end of the table. Reading is always from the first UD in the
  6. ** table and underrun removes the first UD to shift a new one into slot 2. (Slot
  7. ** 1 of the table is reserved for the pipe reader function with 0 denoting no
  8. ** reader.)
  9. **
  10. ** Reads and writes may span multiple UD buffers and if the read spans multiple
  11. ** UDs then the parts are collected as strings on the Lua stack and then
  12. ** concatenated with a lua_concat().
  13. **
  14. ** Note that pipe tables also support the undocumented length and tostring
  15. ** operators for debugging puposes, so if p is a pipe then #p[i] gives the
  16. ** effective length of pipe slot i and printing p[i] gives its contents.
  17. **
  18. ** The pipe library also supports the automatic scheduling of a reader task.
  19. ** This is declared by including a Lua CB function and an optional prioirty for
  20. ** it to execute at in the pipe.create() call. The reader task may or may not
  21. ** empty the FIFO (and there is also nothing to stop the task also writing to
  22. ** the FIFO. The reader automatically reschedules itself if the pipe contains
  23. ** unread content.
  24. **
  25. ** The reader tasks may be interleaved with other tasks that write to the pipe
  26. ** and others that don't. Any task writing to the pipe will also trigger the
  27. ** posting of a read task if one is not already pending. In this way at most
  28. ** only one pending reader task is pending, and this prevents overrun of the
  29. ** task queueing system.
  30. **
  31. ** Implementation Notes:
  32. **
  33. ** - The Pipe slot 1 is used to store the Lua CB function reference of the
  34. ** reader task. Note that is actually an auxiliary wrapper around the
  35. ** supplied Lua CB function, and this wrapper also uses upvals to store
  36. ** internal pipe state. The remaining slots are the Userdata buffer chunks.
  37. **
  38. ** - This internal state needs to be shared with the pipe_write function, but a
  39. ** limitation of Lua 5.1 is that C functions cannot share upvals; to avoid
  40. ** this constraint, this function is also denormalised to act as the
  41. ** pipe_write function: if Arg1 is the pipe then its a pipe:write() otherwise
  42. ** its a CB wrapper.
  43. **
  44. ** Also note that the pipe module is used by the Lua VM and therefore the create
  45. ** read, and unread methods are exposed as directly callable C functions. (Write
  46. ** is available through pipe[1].)
  47. **
  48. ** Read the docs/modules/pipe.md documentation for a functional description.
  49. */
  50. #include "module.h"
  51. #include "lauxlib.h"
  52. #include <string.h>
  53. #include "platform.h"
  54. #define INVALID_LEN ((unsigned)-1)
  55. #define LUA_USE_MODULES_PIPE
  56. typedef struct buffer {
  57. int start, end;
  58. char buf[LUAL_BUFFERSIZE];
  59. } buffer_t;
  60. #define AT_TAIL 0x00
  61. #define AT_HEAD 0x01
  62. #define WRITING 0x02
  63. static buffer_t *checkPipeUD (lua_State *L, int ndx);
  64. static buffer_t *newPipeUD(lua_State *L, int ndx, int n);
  65. static int pipe_write_aux(lua_State *L);
  66. LROT_TABLE(pipe_meta);
  67. /* Validation and utility functions */
  68. // [-0, +0, v]
  69. static buffer_t *checkPipeTable (lua_State *L, int tbl, int flags) {
  70. int m = lua_gettop(L), n = lua_objlen(L, tbl);
  71. if (lua_istable(L, tbl) && lua_getmetatable(L, tbl)) {
  72. lua_pushrotable(L, LROT_TABLEREF(pipe_meta));/* push comparison metatable */
  73. if (lua_rawequal(L, -1, -2)) { /* check these match */
  74. buffer_t *ud;
  75. if (n == 1) {
  76. ud = (flags & WRITING) ? newPipeUD(L, tbl, 2) : NULL;
  77. } else {
  78. int i = flags & AT_HEAD ? 2 : n; /* point to head or tail of T */
  79. lua_rawgeti(L, tbl, i); /* and fetch UD */
  80. ud = checkPipeUD(L, -1);
  81. }
  82. lua_settop(L, m);
  83. return ud; /* and return ptr to buffer_t rec */
  84. }
  85. }
  86. luaL_argerror(L, tbl, "pipe table");
  87. return NULL; /* NORETURN avoid compiler error */
  88. }
  89. static buffer_t *checkPipeUD (lua_State *L, int ndx) { // [-0, +0, v]
  90. luaL_checktype(L, ndx, LUA_TUSERDATA); /* NORETURN on error */
  91. buffer_t *ud = lua_touserdata(L, ndx);
  92. if (ud && lua_getmetatable(L, ndx)) { /* Get UD and its metatable? */
  93. lua_pushrotable(L, LROT_TABLEREF(pipe_meta)); /* push correct metatable */
  94. if (lua_rawequal(L, -1, -2)) { /* Do these match? */
  95. lua_pop(L, 2); /* remove both metatables */
  96. return ud; /* and return ptr to buffer_t rec */
  97. }
  98. }
  99. }
  100. /* Create new buffer chunk at `n` in the table which is at stack `ndx` */
  101. static buffer_t *newPipeUD(lua_State *L, int ndx, int n) { // [-0,+0,-]
  102. buffer_t *ud = (buffer_t *) lua_newuserdata(L, sizeof(buffer_t));
  103. lua_pushrotable(L, LROT_TABLEREF(pipe_meta)); /* push the metatable */
  104. lua_setmetatable(L, -2); /* set UD's metabtable to pipe_meta */
  105. ud->start = ud->end = 0;
  106. lua_rawseti(L, ndx, n); /* T[n] = new UD */
  107. return ud; /* ud points to new T[n] */
  108. }
  109. #define CHAR_DELIM -1
  110. #define CHAR_DELIM_KEEP -2
  111. static char getsize_delim (lua_State *L, int ndx, int *len) { // [-0, +0, v]
  112. char delim;
  113. int n;
  114. if( lua_type( L, ndx ) == LUA_TNUMBER ) {
  115. n = luaL_checkinteger( L, ndx );
  116. luaL_argcheck(L, n > 0, ndx, "invalid chunk size");
  117. delim = '\0';
  118. } else if (lua_isnil(L, ndx)) {
  119. n = CHAR_DELIM;
  120. delim = '\n';
  121. } else {
  122. size_t ldelim;
  123. const char *s = lua_tolstring( L, 2, &ldelim);
  124. n = ldelim == 2 && s[1] == '+' ? CHAR_DELIM_KEEP : CHAR_DELIM ;
  125. luaL_argcheck(L, ldelim + n == 0, ndx, "invalid delimiter");
  126. delim = s[0];
  127. }
  128. if (len)
  129. *len = n;
  130. return delim;
  131. }
  132. /*
  133. ** Read CB Initiator AND pipe_write. If arg1 == the pipe, then this is a pipe
  134. ** write(); otherwise it is the Lua CB wapper for the task post. This botch
  135. ** allows these two functions to share Upvals within the Lua 5.1 VM:
  136. */
  137. #define UVpipe lua_upvalueindex(1) // The pipe table object
  138. #define UVfunc lua_upvalueindex(2) // The CB's Lua function
  139. #define UVprio lua_upvalueindex(3) // The task priority
  140. #define UVstate lua_upvalueindex(4) // Pipe state;
  141. #define CB_NOT_USED 0
  142. #define CB_ACTIVE 1
  143. #define CB_WRITE_UPDATED 2
  144. #define CB_QUIESCENT 4
  145. /*
  146. ** Note that nothing precludes the Lua CB function from itself writing to the
  147. ** pipe and in this case this routine will call itself recursively.
  148. **
  149. ** The Lua CB itself takes the pipe object as a parameter and returns an
  150. ** optional boolean to force or to suppress automatic retasking if needed. If
  151. ** omitted, then the default is to repost if the pipe is not empty, otherwise
  152. ** the task chain is left to lapse.
  153. */
  154. static int pipe_write_and_read_poster (lua_State *L) {
  155. int state = lua_tointeger(L, UVstate);
  156. if (lua_rawequal(L, 1, UVpipe)) {
  157. /* arg1 == the pipe, so this was invoked as a pipe_write() */
  158. if (pipe_write_aux(L) && state && !(state & CB_WRITE_UPDATED)) {
  159. /*
  160. * if this resulted in a write and not already in a CB and not already
  161. * toggled the write update then post the task
  162. */
  163. state |= CB_WRITE_UPDATED;
  164. lua_pushinteger(L, state);
  165. lua_replace(L, UVstate); /* Set CB state write updated flag */
  166. if (state == CB_QUIESCENT | CB_WRITE_UPDATED) {
  167. lua_rawgeti(L, 1, 1); /* Get CB ref from pipe[1] */
  168. luaL_posttask(L, (int) lua_tointeger(L, UVprio)); /* and repost task */
  169. }
  170. }
  171. } else if (state != CB_NOT_USED) {
  172. /* invoked by the luaN_taskpost() so call the Lua CB */
  173. int repost; /* can take the values CB_WRITE_UPDATED or 0 */
  174. lua_pushinteger(L, CB_ACTIVE); /* CB state set to active only */
  175. lua_replace(L, UVstate);
  176. lua_pushvalue(L, UVfunc); /* Lua CB function */
  177. lua_pushvalue(L, UVpipe); /* pipe table */
  178. lua_call(L, 1, 1); /* Errors are thrown back to caller */
  179. /*
  180. * On return from the Lua CB, the task is never reposted if the pipe is empty.
  181. * If it is not empty then the Lua CB return status determines when reposting
  182. * occurs:
  183. * - true = repost
  184. * - false = don't repost
  185. * - nil = only repost if there has been a write update.
  186. */
  187. if (lua_isboolean(L,-1)) {
  188. repost = (lua_toboolean(L, -1) == true &&
  189. lua_objlen(L, UVpipe) > 1) ? CB_WRITE_UPDATED : 0;
  190. } else {
  191. repost = state & CB_WRITE_UPDATED;
  192. }
  193. state = CB_QUIESCENT | repost;
  194. lua_pushinteger(L, state); /* Update the CB state */
  195. lua_replace(L, UVstate);
  196. if (repost) {
  197. lua_rawgeti(L, UVpipe, 1); /* Get CB ref from pipe[1] */
  198. luaL_posttask(L, (int) lua_tointeger(L, UVprio)); /* and repost task */
  199. }
  200. }
  201. return 0;
  202. }
  203. /* Lua callable methods. Since the metatable is linked to both the pipe table */
  204. /* and the userdata entries the __len & __tostring functions must handle both */
  205. // Lua: buf = pipe.create()
  206. int pipe_create(lua_State *L) {
  207. int prio = -1;
  208. lua_settop(L, 2); /* fix stack sze as 2 */
  209. if (!lua_isnil(L, 1)) {
  210. luaL_checktype(L, 1, LUA_TFUNCTION); /* non-nil arg1 must be a function */
  211. if (lua_isnil(L, 2)) {
  212. prio = PLATFORM_TASK_PRIORITY_MEDIUM;
  213. } else {
  214. prio = (int) lua_tointeger(L, 2);
  215. luaL_argcheck(L, prio >= PLATFORM_TASK_PRIORITY_LOW &&
  216. prio <= PLATFORM_TASK_PRIORITY_HIGH, 2,
  217. "invalid priority");
  218. }
  219. }
  220. lua_createtable (L, 1, 0); /* create pipe table */
  221. lua_pushrotable(L, LROT_TABLEREF(pipe_meta));
  222. lua_setmetatable(L, -2); /* set pipe table's metabtable to pipe_meta */
  223. lua_pushvalue(L, -1); /* UV1: pipe object */
  224. lua_pushvalue(L, 1); /* UV2: CB function */
  225. lua_pushinteger(L, prio); /* UV3: task priority */
  226. lua_pushinteger(L, prio == -1 ? CB_NOT_USED : CB_QUIESCENT);
  227. lua_pushcclosure(L, pipe_write_and_read_poster, 4); /* aux func for C task */
  228. lua_rawseti(L, -2, 1); /* and write to T[1] */
  229. return 1; /* return the table */
  230. }
  231. // len = #pipeobj[i]
  232. static int pipe__len (lua_State *L) {
  233. if (lua_istable(L, 1)) {
  234. lua_pushinteger(L, lua_objlen(L, 1));
  235. } else {
  236. buffer_t *ud = checkPipeUD(L, 1);
  237. lua_pushinteger(L, ud->end - ud->start);
  238. }
  239. return 1;
  240. }
  241. //Lua s = pipeUD:tostring()
  242. static int pipe__tostring (lua_State *L) {
  243. if (lua_istable(L, 1)) {
  244. lua_pushfstring(L, "Pipe: %p", lua_topointer(L, 1));
  245. } else {
  246. buffer_t *ud = checkPipeUD(L, 1);
  247. lua_pushlstring(L, ud->buf + ud->start, ud->end - ud->start);
  248. }
  249. return 1;
  250. }
  251. // Lua: rec = p:read(end_or_delim) // also [-2, +1,- ]
  252. int pipe_read(lua_State *L) {
  253. buffer_t *ud = checkPipeTable(L, 1, AT_HEAD);
  254. int i, k=0, n;
  255. lua_settop(L,2);
  256. const char delim = getsize_delim(L, 2, &n);
  257. lua_pop(L,1);
  258. while (ud && n) {
  259. int want, used, avail = ud->end - ud->start;
  260. if (n < 0 /* one of the CHAR_DELIM flags */) {
  261. /* getting a delimited chunk so scan for delimiter */
  262. for (i = ud->start; i < ud->end && ud->buf[i] != delim; i++) {}
  263. /* Can't have i = ud->end and ud->buf[i] == delim so either */
  264. /* we've scanned full buffer avail OR we've hit a delim char */
  265. if (i == ud->end) {
  266. want = used = avail; /* case where we've scanned without a hit */
  267. } else {
  268. want = used = i + 1 - ud->start; /* case where we've hit a delim */
  269. if (n == CHAR_DELIM)
  270. want--;
  271. n = 0; /* force loop exit because delim found */
  272. }
  273. } else {
  274. want = used = (n < avail) ? n : avail;
  275. n -= used;
  276. }
  277. lua_pushlstring(L, ud->buf + ud->start, want); /* part we want */
  278. k++;
  279. ud->start += used;
  280. if (ud->start == ud->end) {
  281. /* shift the pipe array down overwriting T[1] */
  282. int nUD = lua_objlen(L, 1);
  283. for (i = 2; i < nUD; i++) { /* for i = 2, nUD-1 */
  284. lua_rawgeti(L, 1, i+1); lua_rawseti(L, 1, i); /* T[i] = T[i+1] */
  285. }
  286. lua_pushnil(L); lua_rawseti(L, 1, nUD--); /* T[n] = nil */
  287. if (nUD>1) {
  288. lua_rawgeti(L, 1, 2);
  289. ud = checkPipeUD(L, -1);
  290. lua_pop(L, 1);
  291. } else {
  292. ud = NULL;
  293. }
  294. }
  295. }
  296. if (k)
  297. lua_concat(L, k);
  298. else
  299. lua_pushnil(L);
  300. return 1;
  301. }
  302. // Lua: buf:unread(some_string)
  303. int pipe_unread(lua_State *L) {
  304. size_t l = INVALID_LEN;
  305. const char *s = lua_tolstring(L, 2, &l);
  306. if (l==0)
  307. return 0;
  308. luaL_argcheck(L, l != INVALID_LEN, 2, "must be a string");
  309. buffer_t *ud = checkPipeTable(L, 1, AT_HEAD | WRITING);
  310. do {
  311. int used = ud->end - ud->start;
  312. int lrem = LUAL_BUFFERSIZE-used;
  313. if (used == LUAL_BUFFERSIZE) {
  314. /* If the current UD is full insert a new UD at T[2] */
  315. int i, nUD = lua_objlen(L, 1);
  316. for (i = nUD; i > 1; i--) { /* for i = nUD,1,-1 */
  317. lua_rawgeti(L, 1, i); lua_rawseti(L, 1, i+1); /* T[i+1] = T[i] */
  318. }
  319. ud = newPipeUD(L, 1, 2);
  320. used = 0; lrem = LUAL_BUFFERSIZE;
  321. /* Filling leftwards; make this chunk "empty but at the right end" */
  322. ud->start = ud->end = LUAL_BUFFERSIZE;
  323. } else if (ud->start < l) {
  324. /* If the unread can't fit it before the start, shift content to end */
  325. memmove(ud->buf + lrem,
  326. ud->buf + ud->start, used); /* must be memmove not cpy */
  327. ud->start = lrem; ud->end = LUAL_BUFFERSIZE;
  328. }
  329. if (l <= (unsigned )lrem)
  330. break;
  331. /* If we're here then the remaining string is strictly longer than the */
  332. /* remaining buffer space; top off the buffer before looping around again */
  333. l -= lrem;
  334. memcpy(ud->buf, s + l, lrem);
  335. ud->start = 0;
  336. } while(1);
  337. /* Copy any residual tail to the UD buffer.
  338. * Here, ud != NULL and 0 <= l <= ud->start */
  339. ud->start -= l;
  340. memcpy(ud->buf + ud->start, s, l);
  341. return 0;
  342. }
  343. // Lua: buf:write(some_string)
  344. static int pipe_write_aux(lua_State *L) {
  345. size_t l = INVALID_LEN;
  346. const char *s = lua_tolstring(L, 2, &l);
  347. //dbg_printf("pipe write(%u): %s", l, s);
  348. if (l==0)
  349. return false;
  350. luaL_argcheck(L, l != INVALID_LEN, 2, "must be a string");
  351. buffer_t *ud = checkPipeTable(L, 1, AT_TAIL | WRITING);
  352. do {
  353. int used = ud->end - ud->start;
  354. if (used == LUAL_BUFFERSIZE) {
  355. /* If the current UD is full insert a new UD at T[end] */
  356. ud = newPipeUD(L, 1, lua_objlen(L, 1)+1);
  357. used = 0;
  358. } else if (LUAL_BUFFERSIZE - ud->end < l) {
  359. /* If the write can't fit it at the end then shift content to the start */
  360. memmove(ud->buf, ud->buf + ud->start, used); /* must be memmove not cpy */
  361. ud->start = 0; ud->end = used;
  362. }
  363. int lrem = LUAL_BUFFERSIZE-used;
  364. if (l <= (unsigned )lrem)
  365. break;
  366. /* If we've got here then the remaining string is longer than the buffer */
  367. /* space left, so top off the buffer before looping around again */
  368. memcpy(ud->buf + ud->end, s, lrem);
  369. ud->end += lrem;
  370. l -= lrem;
  371. s += lrem;
  372. } while(1);
  373. /* Copy any residual tail to the UD buffer. Note that this is l>0 and */
  374. memcpy(ud->buf + ud->end, s, l);
  375. ud->end += l;
  376. return true;
  377. }
  378. // Lua: fread = pobj:reader(1400) -- or other number
  379. // fread = pobj:reader('\n') -- or other delimiter (delim is stripped)
  380. // fread = pobj:reader('\n+') -- or other delimiter (delim is retained)
  381. #define TBL lua_upvalueindex(1)
  382. #define N lua_upvalueindex(2)
  383. static int pipe_read_aux(lua_State *L) {
  384. lua_settop(L, 0); /* ignore args since we use upvals instead */
  385. lua_pushvalue(L, TBL);
  386. lua_pushvalue(L, N);
  387. return pipe_read(L);
  388. }
  389. static int pipe_reader(lua_State *L) {
  390. lua_settop(L,2);
  391. checkPipeTable (L, 1,AT_HEAD);
  392. getsize_delim(L, 2, NULL);
  393. lua_pushcclosure(L, pipe_read_aux, 2); /* return (closure, nil, table) */
  394. return 1;
  395. }
  396. // return number of records
  397. static int pipe_nrec (lua_State *L) {
  398. lua_pushinteger(L, lua_objlen(L, 1) - 1);
  399. return 1;
  400. }
  401. LROT_BEGIN(pipe_funcs, NULL, 0)
  402. LROT_FUNCENTRY( __len, pipe__len )
  403. LROT_FUNCENTRY( __tostring, pipe__tostring )
  404. LROT_FUNCENTRY( read, pipe_read )
  405. LROT_FUNCENTRY( reader, pipe_reader )
  406. LROT_FUNCENTRY( unread, pipe_unread )
  407. LROT_FUNCENTRY( nrec, pipe_nrec )
  408. LROT_END(pipe_funcs, NULL, 0)
  409. /* Using a index func is needed because the write method is at pipe[1] */
  410. static int pipe__index(lua_State *L) {
  411. lua_settop(L,2);
  412. const char *k=lua_tostring(L,2);
  413. if(!strcmp(k,"write")){
  414. lua_rawgeti(L, 1, 1);
  415. } else {
  416. lua_pushrotable(L, LROT_TABLEREF(pipe_funcs));
  417. lua_replace(L, 1);
  418. lua_rawget(L, 1);
  419. }
  420. return 1;
  421. }
  422. LROT_BEGIN(pipe_meta, NULL, LROT_MASK_INDEX)
  423. LROT_FUNCENTRY( __index, pipe__index)
  424. LROT_FUNCENTRY( __len, pipe__len )
  425. LROT_FUNCENTRY( __tostring, pipe__tostring )
  426. LROT_END(pipe_meta, NULL, LROT_MASK_INDEX)
  427. LROT_BEGIN(pipe, NULL, 0)
  428. LROT_FUNCENTRY( create, pipe_create )
  429. LROT_END(pipe, NULL, 0)
  430. NODEMCU_MODULE(PIPE, "pipe", pipe, NULL);