def write.loop

mylsm-lsm-store@0.6.0.1 · src/coordinator/Io.bend:305

import Base
import mylsm-lsm-store@0.6.0.1/src/coordinator/Protocol.bend as Protocol

305def write.loop(
306  fuel: Nat,
307  state: WriteState
308) -> IO(
309  File & Result<&1, &1, U32 & String, Unit>
310):
311  match fuel:
312    case 0n:
313      match state:
314        case WriteNeed{file, _, _, _, _}:
315          IO.pure(File & Result<&1, &1, U32 & String, Unit>,
316            (file, Fail{(7, "socket write exhausted its progress bound")}))
317        case WriteSent{pair, _, _, _}:
318          match pair:
319            case (file, _):
320              IO.pure(File & Result<&1, &1, U32 & String, Unit>,
321                (file, Fail{(7, "socket write exhausted its progress bound")}))
322        case WriteAdvance{file, _, _, _, _, _}:
323          IO.pure(File & Result<&1, &1, U32 & String, Unit>,
324            (file, Fail{(7, "socket write exhausted its progress bound")}))
325        case WriteFailed{file}:
326          IO.pure(File & Result<&1, &1, U32 & String, Unit>,
327            (file, Fail{(5, "socket write made no progress")}))
328        case WriteDone{file}:
329          IO.pure(File & Result<&1, &1, U32 & String, Unit>,
330            (file, Done{Unit{}}))
331        case WriteSend{file, _, _, _, _}:
332          IO.pure(File & Result<&1, &1, U32 & String, Unit>,
333            (file, Fail{(7, "socket write exhausted its progress bound")}))
334    case 1n+rest:
335      match state:
336        case WriteNeed{file, remaining, total, offsets, bytes}:
337          match remaining:
338            case (remaining_check, (remaining_send, _)):
339              write.loop(rest, write.need(U32.is_eq(remaining_check, 0), file,
340                remaining_send, total, offsets, bytes))
341        case WriteDone{file}:
342          IO.pure(File & Result<&1, &1, U32 & String, Unit>,
343            (file, Done{Unit{}}))
344        case WriteFailed{file}:
345          IO.pure(File & Result<&1, &1, U32 & String, Unit>,
346            (file, Fail{(5, "socket write made no progress")}))
347        case WriteSend{file, total, offset, remaining, words}:
348          match offset:
349            case (offset_send, (offset_keep, _)):
350              match remaining:
351                case (remaining_send, (remaining_keep, _)):
352                  do IO<File & Result<&1, &1, U32 & String, Unit>>:
353                    pair : File & Result<&1, &1, U32 & String, U32 & Array<U32>> <-
354                      write.words(file, offset_send,
355                        write.chunk(remaining_send), words)
356                    write.loop(rest, WriteSent{pair, total, offset_keep,
357                      remaining_keep})
358        case WriteSent{pair, total, offset, remaining}:
359          match pair:
360            case (file, Fail{error}):
361              IO.pure(File & Result<&1, &1, U32 & String, Unit>,
362                (file, Fail{error}))
363            case (file, Done{(sent, words)}):
364              write.loop(rest, WriteAdvance{file, total, offset, remaining,
365                Protocol.u32.duplicate.triple(sent), words})
366        case WriteAdvance{file, total, offset, remaining, copies, words}:
367          match copies:
368            case (sent_check, (sent_offset, sent_remaining)):
369              write.loop(rest, write.advance(U32.is_eq(sent_check, 0),
370                file, total, offset, remaining, sent_offset, sent_remaining,
371                words))