silk.io.buffered_duplex.

When to use#

Use withBuffered or withBufferedCapacity to run a protocol over a connection: read requests, write responses, and close the connection at the end. Use withBufferedPairCapacity to move data between two connections, for example in a proxy with silk.io.buffered_transfer.

Details#

The with... functions allocate an input buffer and an output buffer, give the callback a BufferedDuplex session, and close the transport when the callback returns, fails, or is canceled. The session cannot escape the callback.

The read operations work like silk.io.buffered_input.BufferedInput. The write operations work like silk.io.buffered_output.BufferedOutput. A transport failure makes the whole session terminal, and each later operation fails with BufferError.Terminal.

Gotchas#

The end of the scope does not flush. Bytes that are still pending are lost. Call BufferedDuplex.flush or BufferedDuplex.shutdownWrite before the callback returns. If the program traps, the transport is not closed.

Examples#

Read a request and write a response#

import silk.mem.allocator { Allocator, OutOfMemoryError }
import silk.io.buffered_duplex { BufferedDuplex, withBufferedCapacity }
import silk.io.buffered_input { BufferError }
import silk.io.byte_duplex { ByteDuplex }
import silk.collections.bytes { Bytes }
import silk.core.effect { Effect }
import silk.collections.list { List }
import silk.io.memory_byte_duplex {
  MemoryByteDuplex,
  MemoryReadEvent,
  MemoryWriteAction,
  MemoryWriteEvent
}
import silk.system.monotonic_clock { MonotonicClock }
import silk.core.option { Option }
import silk.system.system_clock { Instant, SystemClock }

struct FixedClock {}

impl MonotonicClock for FixedClock {
  effect fn now(self: &mut Self) -> Instant {
    return SystemClock.make(0, 0)
  }

  effect fn getResolution(self: &mut Self) -> u64 {
    return 1
  }

  effect fn waitUntil(self: &mut Self, when: Instant) -> () {
    return ()
  }

  effect fn waitFor(self: &mut Self, howLong: u64) -> () {
    return ()
  }
}

effect fn transport() -> MemoryByteDuplex
  ! OutOfMemoryError
  ? &mut Allocator {
  let mut reads = List<MemoryReadEvent>.make()
  let bytes = run Bytes.copy(&b"ping")
  run List<MemoryReadEvent>.append(
    &mut reads,
    MemoryReadEvent.Data { readyAt: SystemClock.make(0, 0), bytes: move bytes }
  )
  let mut writes = List<MemoryWriteEvent>.make()
  run List<MemoryWriteEvent>.append(
    &mut writes,
    MemoryWriteEvent {
      readyAt: SystemClock.make(0, 0),
      action: MemoryWriteAction.Accept { count: 4 }
    }
  )
  return run MemoryByteDuplex.make(move reads, move writes, 4, 8, Option<i32>.none())
}

// Reads a 4-byte request and answers it.
effect<'session> fn answer<'session, P>(session: &'session mut BufferedDuplex<'session, P>) -> ()
  ! BufferError
  ? &mut MonotonicClock where
  &'session mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock {
  let mut request: [u8; 4] = [0, 0, 0, 0]
  run BufferedDuplex.readExact(&mut session.*, &mut request, Option<Instant>.none())
  run BufferedDuplex.writeAll(&mut session.*, b"pong", Option<Instant>.none())
  return run BufferedDuplex.flush(&mut session.*, Option<Instant>.none())
}

pub effect fn main() -> i32
  ! BufferError | OutOfMemoryError {
  let mut allocator = Allocator.systemAllocatorProvider()
  let mut clock = FixedClock {}
  let mut duplex = run transport() |> Effect.provideMut<Allocator>(&mut allocator)
  run withBufferedCapacity<(), BufferError>(&mut duplex, 16, 16, answer)
    |> Effect.provideMut<MonotonicClock>(&mut clock)
    |> Effect.provideMut<Allocator>(&mut allocator)
  let sent = MemoryByteDuplex.outbound(&duplex)
  if sent.length != 4 || sent[0] != 112 || sent[1] != 111 {
    return 1
  }
  return 0
}

Import as BufferedDuplex with import silk.io.buffered_duplex { BufferedDuplex }.

Public declarations: 7.

DEFAULT_CAPACITY#

pub const DEFAULT_CAPACITY: usize

The capacity of each buffer in withBuffered: 8 KiB (8,192 bytes).

BufferedDuplex#

pub struct BufferedDuplex<'transport, P>

A buffered session over one transport of type P. The with... functions make it and pass it to the callback.

Details#

The session has an input buffer and an output buffer. It borrows the transport for the duration of the callback and cannot escape it. The scope closes the transport when the callback returns, fails, or is canceled. A transport failure makes the whole session terminal, and each later operation fails with BufferError.Terminal.

Gotchas#

The end of the scope does not flush. Bytes that are still pending are lost. Call BufferedDuplex.flush or BufferedDuplex.shutdownWrite before the callback returns. If the program traps, the transport is not closed.

BufferedContext#

pub interface BufferedContext<P, A, E, ?R>

A value that can use a buffered session, as an alternative to a callback.

When to use#

Implement it for a type that owns state that can be used only one time, and pass the value to withBufferedCapacityContext. Use a callback with withBufferedCapacity in other cases.

Details#

use consumes the context and receives the session. The session cannot escape through the context or the result.

Operation use#

effect<'session> fn use<'session, 'transport: 'session>(
    context: Self,
    session: &'session mut BufferedDuplex<'transport, P>,
  ) -> A ! E ? R
  where &'transport mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Consumes context and uses session.

Implementation BufferedDuplex<'transport, P>#

impl<'transport, P> BufferedDuplex<'transport, P>

Method peek#

pub fn peek<'a>(self: &'a Self) -> &'a [u8]

Returns the unread input bytes. Returns an empty slice after BufferedDuplex.close.

Method unread#

pub fn unread(self: &Self) -> usize

Returns the number of unread input bytes. Returns 0 after BufferedDuplex.close.

Method pending#

pub fn pending(self: &Self) -> usize

Returns the number of output bytes that the transport has not received. Returns 0 after BufferedDuplex.close.

Method consume#

pub effect fn consume(self: &mut Self, count: usize) -> () ! BufferError

Removes count unread input bytes from the front of the input buffer.

Errors#

• BufferError.Terminal if an earlier failure or a close made the session unusable.

• BufferError.InvalidConsumption if count is more than BufferedDuplex.unread.

Method fill#

pub effect fn fill(
    self: &mut Self,
    minimum: usize,
    deadline: Option<Instant>,
  ) -> FillOutcome
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Reads until the input buffer holds at least minimum unread bytes, or until the transport ends.

Details#

If the input buffer already holds minimum bytes, it does not read. Each read asks for all free space, so the buffer can hold more than minimum bytes after the call. A minimum of 0 never reads. deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit.

Errors#

• BufferError.Terminal if an earlier failure or a close made the session unusable.

• BufferError.BufferTooSmall if minimum is more than the input capacity.

• BufferError.ReadFailed if a read fails. The session becomes terminal. progress gives the bytes that this call added.

Method readSome#

pub effect fn readSome(
    self: &mut Self,
    output: &mut [u8],
    deadline: Option<Instant>,
  ) -> ReadTransfer
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Copies unread bytes into output and consumes them. Reads first if the input buffer is empty.

Details#

The result is ReadTransfer.Data with a count from 1 to output.length, or ReadTransfer.End if the transport ended and the input buffer is empty. An empty output returns a count of 0 and does not read. deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit.

Errors#

• BufferError.Terminal if an earlier failure or a close made the session unusable.

• BufferError.ReadFailed if a read fails. The session becomes terminal. progress gives the bytes that this call added.

Method readExact#

pub effect fn readExact(
    self: &mut Self,
    output: &mut [u8],
    deadline: Option<Instant>,
  ) -> ()
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Fills all of output with the next input bytes and consumes them.

Details#

deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit. It applies to each read.

Errors#

• BufferError.Terminal if an earlier failure or a close made the session unusable.

• BufferError.UnexpectedEnd if the transport ends first. progress gives the number of bytes that were copied into output.

• BufferError.ReadFailed if a read fails. The session becomes terminal. progress includes the bytes that this call copied before the failure.

Method discardAtMost#

pub effect fn discardAtMost(
    self: &mut Self,
    limit: usize,
    deadline: Option<Instant>,
  ) -> DiscardOutcome
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Consumes and drops up to limit input bytes. It stops early if the transport ends.

Details#

The result gives the number of discarded bytes and if the transport ended. deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit.

Errors#

• BufferError.Terminal if an earlier failure or a close made the session unusable.

• BufferError.ReadFailed if a read fails. The session becomes terminal. progress gives the bytes that this call added.

Method discardExact#

pub effect fn discardExact(
    self: &mut Self,
    count: usize,
    deadline: Option<Instant>,
  ) -> ()
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Consumes and drops exactly count input bytes.

Details#

deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit.

Errors#

• BufferError.Terminal if an earlier failure or a close made the session unusable.

• BufferError.UnexpectedEnd if the transport ends first. progress gives the number of bytes that were discarded.

• BufferError.ReadFailed if a read fails. The session becomes terminal. progress gives the bytes that this call added.

Method writeSome#

pub effect fn writeSome(
    self: &mut Self,
    input: &[u8],
    deadline: Option<Instant>,
  ) -> usize
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Copies a prefix of input into the output buffer and returns its length. Sends pending bytes first if the buffer is full.

Details#

If input is longer than the output capacity, it does not copy: it sends the pending bytes and input to the transport until the transport accepts some input bytes. For a non-empty input, the result is at least 1. For an empty input, the result is 0 and the transport is not used. deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit.

Errors#

• BufferError.Terminal if an earlier failure, a close, or BufferedDuplex.shutdownWrite made output unusable.

• BufferError.WriteFailed if the transport fails or deadline passes. The session becomes terminal. drained gives the number of pending bytes that the transport received.

Method writeAll#

pub effect fn writeAll(
    self: &mut Self,
    input: &[u8],
    deadline: Option<Instant>,
  ) -> ()
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Copies all of input into the output buffer, and sends pending bytes as the buffer fills.

Details#

When it returns, the output buffer owns all of input, or the transport already has them. The transport does not necessarily have the buffered bytes yet. Call BufferedDuplex.flush to send them. deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit. It applies to each transport write.

Errors#

• BufferError.Terminal if an earlier failure, a close, or BufferedDuplex.shutdownWrite made output unusable.

• BufferError.WriteFailed if the transport fails or deadline passes. The session becomes terminal. accepted gives the number of input bytes that the buffer accepted before the failure. drained gives the number of pending bytes that the transport received.

Method writeVecAll#

pub effect fn writeVecAll(
    self: &mut Self,
    inputs: &[Bytes],
    deadline: Option<Instant>,
  ) -> ()
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Copies each slice in inputs into the output buffer, in order.

Details#

Before it writes, it adds the lengths of all elements and checks the sum for overflow. The transport does not necessarily have the buffered bytes when it returns. deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit.

Errors#

• BufferError.Terminal if an earlier failure, a close, or BufferedDuplex.shutdownWrite made output unusable.

• BufferError.LengthOverflow if the sum of the element lengths overflows usize. Nothing is written.

• BufferError.WriteFailed if the transport fails or deadline passes. The session becomes terminal. accepted counts the accepted bytes of all elements together.

Method flush#

pub effect fn flush(
    self: &mut Self,
    deadline: Option<Instant>,
  ) -> ()
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Sends all pending bytes to the transport and then flushes the transport.

Details#

It sends pending bytes in order, one time each. When it returns, BufferedDuplex.pending is 0. deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit. It applies to each transport operation.

Errors#

• BufferError.Terminal if an earlier failure, a close, or BufferedDuplex.shutdownWrite made output unusable.

• BufferError.WriteFailed if a transport write or the transport flush fails, or if deadline passes. The session becomes terminal. drained gives the number of pending bytes that the transport received.

Method finish#

pub effect fn finish(
    self: &mut Self,
    deadline: Option<Instant>,
  ) -> ()
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Sends all pending bytes to the transport and then flushes the transport.

When to use#

Use it as the last write operation of the session. The end of the scope does not flush. BufferedDuplex.flush has the same behavior.

Details#

It sends pending bytes in order, one time each. When it returns, BufferedDuplex.pending is 0. deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit. It applies to each transport operation.

Errors#

• BufferError.Terminal if an earlier failure, a close, or BufferedDuplex.shutdownWrite made output unusable.

• BufferError.WriteFailed if a transport write or the transport flush fails, or if deadline passes. The session becomes terminal. drained gives the number of pending bytes that the transport received.

Method shutdownWrite#

pub effect fn shutdownWrite(
    self: &mut Self,
    deadline: Option<Instant>,
  ) -> ()
  ! BufferError
  ? &mut MonotonicClock
  where &mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock

Sends all pending bytes, flushes the transport, and then closes the write direction of the transport.

When to use#

Use it to tell the peer that no more data comes, while you still read its answer.

Details#

After it succeeds, reads still work, and writes fail with BufferError.Terminal. A second call does nothing. deadline is an absolute time on the active MonotonicClock. Option.none() means no time limit.

Errors#

• BufferError.Terminal if an earlier failure or a close made the session unusable.

• BufferError.WriteFailed if the flush or the shutdown fails. The session becomes terminal.

Method close#

pub effect fn close(self: &mut Self) -> () ! ByteIoError
  where &mut P provides &ByteDuplex from &mut ByteDuplex

Closes the transport now and drops all buffered bytes. It does not flush.

Details#

The session becomes terminal before the transport closes. After the call starts, BufferedDuplex.peek, BufferedDuplex.unread, and BufferedDuplex.pending report no bytes. A later call does nothing and succeeds, also if the first call failed. The scope does not close the transport again.

Errors#

• ByteIoError if the transport reports a failure when it closes.

withBuffered#

pub effect fn withBuffered<'env, A, E, ?CallbackRequirements, P>(
  transport: &'env mut P,
  callback: for<'call> once fn<'env>(
    &'call mut BufferedDuplex<'call, P>
  ) -> once Effect<'call & 'env; A ! E ? CallbackRequirements>,
) -> A
! E | BufferError | OutOfMemoryError
? CallbackRequirements | &mut Allocator
where &'env mut P provides &ByteDuplex from &mut ByteDuplex,
  &'env mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock,
  CallbackRequirements in Without<CallbackRequirements, ByteDuplex>

Runs callback with a buffered session over transport, with DEFAULT_CAPACITY for each buffer, and closes the transport at the end.

Details#

The transport is closed one time, after the callback returns, fails, or is canceled. A failure of that close is ignored.

Errors#

• BufferError.InvalidCapacity if a capacity is 0 or more than MAX_CAPACITY. Nothing is allocated and the transport is not used.

• OutOfMemoryError if the Allocator cannot supply the buffers.

• The errors of the callback.

Gotchas#

The scope closes the transport at the end but does not flush. Flush in the callback.

withBufferedCapacity#

pub effect fn withBufferedCapacity<'env, A, E, ?CallbackRequirements, P>(
  transport: &'env mut P,
  inputCapacity: usize,
  outputCapacity: usize,
  callback: for<'call> once fn<'env>(
    &'call mut BufferedDuplex<'call, P>
  ) -> once Effect<'call & 'env; A ! E ? CallbackRequirements>,
) -> A
! E | BufferError | OutOfMemoryError
? CallbackRequirements | &mut Allocator
where &'env mut P provides &ByteDuplex from &mut ByteDuplex,
  &'env mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock,
  CallbackRequirements in Without<CallbackRequirements, ByteDuplex>

Runs callback with a buffered session over transport, with the given buffer capacities, and closes the transport at the end.

Details#

The capacities are checked before anything is allocated. The transport is closed one time, after the callback returns, fails, or is canceled. A failure of that close is ignored.

Errors#

• BufferError.InvalidCapacity if a capacity is 0 or more than MAX_CAPACITY. Nothing is allocated and the transport is not used.

• OutOfMemoryError if the Allocator cannot supply the buffers.

• The errors of the callback.

Gotchas#

The scope closes the transport at the end but does not flush. Flush in the callback.

withBufferedCapacityContext#

pub effect<'env> fn withBufferedCapacityContext<
  'env,
  A,
  E,
  ?CallbackRequirements,
  P,
  C: BufferedContext<P, A, E ? CallbackRequirements> + 'env,
>(
  transport: &'env mut P,
  inputCapacity: usize,
  outputCapacity: usize,
  context: C,
) -> A
! E | BufferError | OutOfMemoryError
? CallbackRequirements | &mut Allocator
where &'env mut P provides &ByteDuplex from &mut ByteDuplex,
  &'env mut P provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock,
  CallbackRequirements in Without<CallbackRequirements, ByteDuplex>

Runs context.use with a buffered session over transport, with the given buffer capacities, and closes the transport at the end.

When to use#

Use it instead of withBufferedCapacity when the code that uses the session owns a value that can be used only one time. Implement BufferedContext for that value. use consumes the context and receives the session. The session cannot escape through the context or the result.

Details#

The capacities are checked before anything is allocated. The transport is closed one time, after context.use returns, fails, or is canceled. A failure of that close is ignored.

Errors#

• BufferError.InvalidCapacity if a capacity is 0 or more than MAX_CAPACITY. Nothing is allocated and the transport is not used.

• OutOfMemoryError if the Allocator cannot supply the buffers.

• The errors of the callback.

Gotchas#

The scope closes the transport at the end but does not flush. Flush in the callback.

withBufferedPairCapacity#

pub effect fn withBufferedPairCapacity<
  'env,
  A,
  E,
  ?CallbackRequirements,
  SP,
  DP,
>(
  source: &'env mut SP,
  sourceInputCapacity: usize,
  sourceOutputCapacity: usize,
  destination: &'env mut DP,
  destinationInputCapacity: usize,
  destinationOutputCapacity: usize,
  callback: for<'sourceCall, 'destinationCall> once fn<'env>(
    &'sourceCall mut BufferedDuplex<'sourceCall, SP>,
    &'destinationCall mut BufferedDuplex<'destinationCall, DP>,
  ) -> once Effect<'sourceCall & 'destinationCall; A ! E ? CallbackRequirements>,
) -> A
! E | BufferError | OutOfMemoryError
? CallbackRequirements | &mut Allocator
where &'env mut SP provides &ByteDuplex from &mut ByteDuplex,
  &'env mut DP provides &ByteDuplex from &mut ByteDuplex,
  &'env mut SP provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock,
  &'env mut DP provides &ByteDuplex from &mut ByteDuplex | &mut MonotonicClock,
  CallbackRequirements in Without<CallbackRequirements, ByteDuplex>

Runs callback with two buffered sessions, one over source and one over destination, and closes both transports at the end.

When to use#

Use it with silk.io.buffered_transfer.BufferedTransfer to copy data from one connection to another.

Details#

All four capacities are checked before anything is allocated. Each transport is closed one time, after the callback returns, fails, or is canceled.

Errors#

• BufferError.InvalidCapacity if a capacity is 0 or more than MAX_CAPACITY. Nothing is allocated and the transport is not used.

• OutOfMemoryError if the Allocator cannot supply the buffers.

• The errors of the callback.

Gotchas#

The scope closes the transport at the end but does not flush. Flush in the callback.

Category: Input and output.

Profiles: aarch64-apple-darwin, aarch64-unknown-linux-gnu, aarch64-unknown-linux-gnu-no-libc, x86_64-unknown-linux-gnu, x86_64-unknown-linux-gnu-no-libc, wasm32-unknown-unknown, wasm32-wasip1, wasm32-wasip2.