fifosock.lua 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134
  1. -- Wrap a two-staged fifo around a socket's send; see
  2. -- docs/lua-modules/fifosock.lua for more documentation.
  3. --
  4. -- See fifosocktest.lua for some examples of use or tricky cases.
  5. --
  6. -- Our fifos can take functions; these can be useful for either lazy
  7. -- generators or callbacks for parts of the stream having been sent.
  8. local BIGTHRESH = 256 -- how big is a "big" string?
  9. local SPLITSLOP = 16 -- any slop in the big question?
  10. local FSMALLLIM = 32 -- maximum number of small strings held
  11. local COALIMIT = 3
  12. local concat = table.concat
  13. local insert = table.insert
  14. local gc = collectgarbage
  15. local function wrap(sock)
  16. -- the two fifos
  17. local fsmall, lsmall, fbig = {}, 0, (require "fifo").new()
  18. -- ssend last aggregation string and aggregate count
  19. local ssla, sslan = nil, 0
  20. local ssend = function(s,islast)
  21. local ns = nil
  22. -- Optimistically, try coalescing FIFO dequeues. But, don't try to
  23. -- coalesce function outputs, since functions might be staging their
  24. -- execution on the send event implied by being called.
  25. if type(s) == "function" then
  26. if sslan ~= 0 then
  27. sock:send(ssla)
  28. ssla, sslan = nil, 0; gc()
  29. return s, false -- stay as is and wait for :on("sent")
  30. end
  31. s, ns = s()
  32. elseif type(s) == "string" and sslan < COALIMIT then
  33. if sslan == 0
  34. then ssla, sslan = s, 1
  35. else ssla, sslan = ssla .. s, sslan + 1
  36. end
  37. if islast then
  38. -- this is shipping; if there's room, steal the small fifo, too
  39. if sslan < COALIMIT then
  40. sock:send(ssla .. concat(fsmall))
  41. fsmall, lsmall = {}, 0
  42. else
  43. sock:send(ssla)
  44. end
  45. ssla, sslan = "", 0; gc()
  46. return nil, false
  47. else
  48. return nil, true
  49. end
  50. end
  51. -- Either that was a function or we've hit our coalescing limit or
  52. -- we didn't ship above. Ship now, if there's something to ship.
  53. if s ~= nil then
  54. if sslan == 0 then sock:send(s) else sock:send(ssla .. s) end
  55. ssla, sslan = nil, 0; gc()
  56. return ns or nil, false
  57. elseif sslan ~= 0 then
  58. assert (ns == nil)
  59. sock:send(ssla)
  60. ssla, sslan = nil, 0; gc()
  61. return nil, false
  62. else
  63. assert (ns == nil)
  64. return nil, true
  65. end
  66. end
  67. -- Move fsmall to fbig; might send if fbig empty
  68. local function promote(f)
  69. if #fsmall == 0 then return end
  70. local str = concat(fsmall)
  71. fsmall, lsmall = {}, 0
  72. fbig:queue(str, f or ssend)
  73. end
  74. local function sendnext()
  75. if not fbig:dequeue(ssend) then promote() end
  76. end
  77. sock:on("sent", sendnext)
  78. return function(s)
  79. -- don't sweat the petty things
  80. if s == nil or s == "" then return end
  81. -- Function? Go ahead and queue this thing in the right place.
  82. if type(s) == "function" then promote(); fbig:queue(s, ssend); return; end
  83. s = tostring(s)
  84. -- cork sending until the end in case we're the head of line
  85. local corked = false
  86. local function corker(t) corked = true; return t end
  87. -- small fifo would overfill? promote it
  88. if lsmall + #s > BIGTHRESH or #fsmall >= FSMALLLIM then promote(corker) end
  89. -- big string? chunk and queue big components immediately
  90. -- behind any promotion that just took place
  91. while #s > BIGTHRESH + SPLITSLOP do
  92. local pfx
  93. pfx, s = s:sub(1,BIGTHRESH), s:sub(BIGTHRESH+1)
  94. fbig:queue(pfx, corker)
  95. end
  96. -- Big string? queue and maybe tx now
  97. if #s > BIGTHRESH then fbig:queue(s, corker)
  98. -- small and fifo in immediate dequeue mode
  99. elseif fbig._go and lsmall == 0 then fbig:queue(s, corker)
  100. -- small and queue already moving; let it linger in the small fifo
  101. else insert(fsmall, s) ; lsmall = lsmall + #s
  102. end
  103. -- if it happened that we corked the transmission above...
  104. -- if we queued a good amount of data, go ahead and start transmitting;
  105. -- otherwise, wait a tick and hopefully we will queue more in the interim
  106. -- before transmitting.
  107. if corked then
  108. if #fbig <= COALIMIT
  109. then tmr.create():alarm(1, tmr.ALARM_SINGLE, sendnext)
  110. else sendnext()
  111. end
  112. end
  113. end
  114. end
  115. return { wrap = wrap }