Go

Start the two in either order, in two terminals (go run server.go, go run client.go; in a checkout of the repository, go run ./examples/server and go run ./examples/client in bindings/go): fipc.ConnectStream waits up to its timeout for a server to listen and accept; fipc.Forever waits as long as it takes. The pair makes the exchange every example on this site makes, the message ping, answered in upper case: this client works with any language's server, and this server with any language's client.

package main

import (
	"bytes"
	"log"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	listener, err := fipc.CreateStream("my_channel", 1<<20) // rings of 1 MiB each way
	if err != nil {
		log.Fatal(err)
	}
	defer listener.Close()
	conn, err := listener.Accept(fipc.Forever) // waits for a client
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	message, err := conn.Receive(fipc.Forever)
	if err != nil {
		log.Fatal(err)
	}
	if err := conn.Send(bytes.ToUpper(message), fipc.Forever); err != nil {
		log.Fatal(err)
	}
}

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	conn, err := fipc.ConnectStream("my_channel", 5*time.Second) // waits up to 5 s for the server
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	if err := conn.Send([]byte("ping"), fipc.Forever); err != nil {
		log.Fatal(err)
	}
	reply, err := conn.Receive(fipc.Forever)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(string(reply)) // PING
}

A connection carries messages: Send takes a []byte of any size (at least 1 byte), and Receive returns each message whole, in a new slice; ReceiveInto fills a buffer of yours and returns the length, without allocating. A message longer than the ring travels in pieces and arrives whole. This server echoes every message back until its client leaves, which ends its next Receive with fipc.ErrDisconnected.

package main

import (
	"errors"
	"fmt"
	"log"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	// rings of 1 MiB each way: a power of two, 1 KiB to 2 GiB
	listener, err := fipc.CreateStream("demo", 1<<20)
	if err != nil {
		log.Fatal(err)
	}
	defer listener.Close()
	conn, err := listener.Accept(fipc.Forever) // waits for a client
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	for {
		msg, err := conn.Receive(fipc.Forever)
		if errors.Is(err, fipc.ErrDisconnected) {
			break
		} else if err != nil {
			log.Fatal(err)
		}
		if err := conn.Send(msg, fipc.Forever); err != nil { // echo, any size
			log.Fatal(err)
		}
	}
	fmt.Println("client gone")
}

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	conn, err := fipc.ConnectStream("demo", 5*time.Second)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	for _, word := range []string{"hello", "shared", "memory"} {
		if err := conn.Send([]byte(word), fipc.Forever); err != nil {
			log.Fatal(err)
		}
		echo, err := conn.Receive(fipc.Forever)
		if err != nil {
			log.Fatal(err)
		}
		fmt.Println(string(echo))
	}
}

A call the library refuses returns one of fipc.ErrTimeout, ErrDisconnected, ErrCancelled, ErrTooLarge, ErrInvalid, ErrNoMemory, ErrAddrInUse, on a broadcast ErrOverrun or ErrUnsubscribed, and on a service's client ErrAbandoned, the C API's results, as *fipc.Error values: test them with errors.Is. Their messages are the C names (FIPC_TIMEOUT: the timeout ran out), and ErrTimeout's Timeout() is true, as for the deadlines of packages os and net. A call on a closed handle returns fipc.ErrClosed; one that can't load the library, an error that wraps fipc.ErrNoLibrary and says where it looked. The timeout comes last, a time.Duration counted in whole milliseconds, rounded up: fipc.NoWait (0) doesn't wait, fipc.Forever (or any negative duration) waits as long as it takes. ErrDisconnected is final: close the connection, then accept or connect a new one.

msg, err := conn.Receive(100 * time.Millisecond)
switch {
case err == nil:
	handle(msg)
case errors.Is(err, fipc.ErrTimeout): // nothing yet
default:
	return err // fipc.ErrDisconnected: the peer is gone
}

n, err := conn.ReceiveInto(buf, fipc.Forever)
var e *fipc.Error
if errors.As(err, &e) && e.Len() > 0 { // ErrTooLarge: make room, then again
	buf = make([]byte, e.Len())
	n, err = conn.ReceiveInto(buf, fipc.Forever)
}

Write a message into the ring and commit it; read one in the ring and release it. SendAcquire(n) returns a slice over room in the ring, valid until SendCommit, the next send or Close; ReceiveAcquire returns a read-only slice over the message, valid until ReceiveRelease, the next receive or Close. These slices point into shared memory: using one after that is undefined and may crash the program, so copy what must outlive it, and keep the connection reachable while you use one. A message of several pieces is ErrTooLarge for ReceiveAcquire and stays queued for Receive. In Go, zero-copy is slower than copying for short messages (below): it takes two calls into the library per message on each side where a copy takes one, and through purego a call costs more than copying a short message. It pays off for long messages you would otherwise copy again.

buf, err := conn.SendAcquire(5, fipc.Forever) // room in the ring
if err != nil {
	return err
}
copy(buf, "hello")
return conn.SendCommit(5) // sent; buf is no longer valid

msg, err := conn.ReceiveAcquire(fipc.Forever) // the message in the ring, read-only
if err != nil {
	return err
}
handle(msg)           // read it in place
conn.ReceiveRelease() // frees its room; msg is no longer valid

A broadcast sends every message of one publisher to each of its subscribers, up to 1024, in any processes of the user; each message is stored once, in memory every subscriber maps. Messages are of one piece, up to a quarter of the capacity (publisher.MaxPiece()). fipc.CreateBroadcast(name, capacity, maxSubscribers, maxStall) returns the publisher, a *Conn whose sends go to every subscriber, the broadcast's one handle. fipc.JoinBroadcast(name, role, timeout) returns the subscriber's *Conn. The publisher's process admits each joiner by itself, on a thread of the library's: no goroutine waits for them. The publisher waits for its subscriber (publisher.WaitForMembers(1, timeout) returns the count once at least one has joined) before it sends tick 1 to tick 3; closing the publisher ends the broadcast and frees the name. The subscriber joins gated (fipc.Gated), so it misses nothing (the publisher waits for it), and receives until fipc.ErrDisconnected, which comes after the last message. A gated one that holds the publisher back for the broadcast's maxStall is dropped (fipc.ErrUnsubscribed, for good); fipc.Forever never drops one, and fipc.NoWait refuses gated subscribers.

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	// 64 KiB of messages, up to 8 subscribers; one that holds the publisher back 5 s is dropped
	publisher, err := fipc.CreateBroadcast("my_broadcast", 1<<16, 8, 5*time.Second)
	if err != nil {
		log.Fatal(err)
	}
	if _, err := publisher.WaitForMembers(1, fipc.Forever); err != nil { // waits for the first subscriber
		log.Fatal(err)
	}
	for i := 1; i <= 3; i++ {
		if err := publisher.Send(fmt.Appendf(nil, "tick %d", i), fipc.Forever); err != nil {
			log.Fatal(err)
		}
	}
	publisher.Close() // ends the broadcast: subscribers get fipc.ErrDisconnected after the ticks
}

package main

import (
	"errors"
	"fmt"
	"log"
	"time"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	// gated: misses nothing (the publisher waits for it)
	subscriber, err := fipc.JoinBroadcast("my_broadcast", fipc.Gated, 5*time.Second)
	if err != nil {
		log.Fatal(err)
	}
	defer subscriber.Close()
	for {
		tick, err := subscriber.Receive(fipc.Forever)
		if errors.Is(err, fipc.ErrDisconnected) {
			return // the publisher closed
		} else if err != nil {
			log.Fatal(err)
		}
		fmt.Println(string(tick)) // tick 1, tick 2, tick 3
	}
}

msg, err := subscriber.Receive(fipc.Forever)
var e *fipc.Error
switch {
case err == nil:
	handle(msg)
case errors.As(err, &e) && e.Lost() > 0: // ErrOverrun: a lossy subscriber fell a lap behind
	log.Printf("%d messages lost", e.Lost()) // the next receive goes on from a recent message
default:
	return err // fipc.ErrDisconnected: the broadcast ended
}

n, err := publisher.WaitForMembers(3, 5*time.Second) // waits up to 5 s for 3 subscribers
if errors.Is(err, fipc.ErrTimeout) {
	log.Printf("only %d of 3 subscribers joined", n)
} else if err != nil {
	return err
}

A work queue takes messages from up to 63 producers and hands each to one of up to 64 consumers, in any processes of the user. Each producer has a ring of its own; a consumer takes a message of any producer: the queue promises no order. Messages are of one piece, up to a quarter of the capacity (MaxPiece()), and wait in the queue until a consumer takes them. Each message goes to one consumer, and again, to another, only if that consumer's process ends before acknowledging it (delivery, below). fipc.CreateQueue(name, capacity, maxProducers, maxConsumers) returns the owner's *fipc.Queue; producers and consumers join with fipc.JoinQueue (fipc.Producer, fipc.Consumer), the owner's own too. The owner's process admits each joiner by itself, on a thread of the library's, so an owner and its members may share a goroutine. Here the owner joins as the producer, sends job 1 to job 3, which wait in the queue, waits for the consumer (queue.WaitForMembers(fipc.Consumer, 1, timeout); fipc.AnyQueueRole counts both roles), then closes its producer and the queue. queue.Close() ends the queue: the consumer receives what is left, then fipc.ErrDisconnected, and a producer's sends return fipc.ErrDisconnected from then on; a producer's or a consumer's Close never ends it. The garbage collector never closes a queue: a queue lives until Close or the end of its process, and dropping the *fipc.Queue doesn't end it. A consumer receives copied (Receive, ReceiveInto) or in place (ReceiveAcquire, until ReceiveRelease).

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	// 4 KiB of messages per producer, 1 producer and 1 consumer: this process owns the queue and is its producer
	queue, err := fipc.CreateQueue("my_queue", 4096, 1, 1)
	if err != nil {
		log.Fatal(err)
	}
	producer, err := fipc.JoinQueue("my_queue", fipc.Producer, 5*time.Second)
	if err != nil {
		log.Fatal(err)
	}
	for i := 1; i <= 3; i++ { // each job waits in the queue until a consumer takes it
		if err := producer.Send(fmt.Appendf(nil, "job %d", i), fipc.Forever); err != nil {
			log.Fatal(err)
		}
	}
	if _, err := queue.WaitForMembers(fipc.Consumer, 1, fipc.Forever); err != nil { // waits for the consumer to join
		log.Fatal(err)
	}
	producer.Close()
	queue.Close() // ends the queue: the consumer gets fipc.ErrDisconnected after the jobs
}

package main

import (
	"errors"
	"fmt"
	"log"
	"time"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	consumer, err := fipc.JoinQueue("my_queue", fipc.Consumer, 5*time.Second)
	if err != nil {
		log.Fatal(err)
	}
	defer consumer.Close()
	for {
		job, err := consumer.Receive(fipc.Forever)
		if errors.Is(err, fipc.ErrDisconnected) {
			return // the queue ended
		} else if err != nil {
			log.Fatal(err)
		}
		fmt.Println(string(job)) // job 1, job 2, job 3
	}
}

Delivery. Each message goes to one consumer. It goes again, to another consumer, only if the process of the consumer that took it ends before acknowledging it, and when a consumer acknowledges is its own choice, not the queue's. A message received in place is acknowledged at its release (or the consumer's next receive, or its close).

• Choose by the work, per consumer. Work that may safely run twice (rendering, a computation, re-indexing, converting a file) acknowledges after it; work whose effects must never repeat (a payment, an email, a motor command) acknowledges before it.
• Not covered: a consumer that hangs while its process runs (there are no timeouts, by design); the owner's process, with which the queue ends; a reboot. A consumer holds at most one unacknowledged message at a time, the one received in place.
• A message that goes again comes in no set order, after messages its producer sent later: the queue promises no order anyway.
• A message held in place holds back its producer's room until its release: a consumer that acknowledges after a long job keeps that message's room, and so the room of the producer's messages after it, until then.
• A message that crashes every consumer that takes it, acknowledged after the work, goes to each consumer in turn, with no limit: acknowledge before the work where that can happen.

A service takes requests from up to 64 clients and hands each to one of up to 64 servers, in any processes of the user; the server that took a request replies to it, and the reply goes back to the client that sent it. A server waits for the requests of every client at once (one Receive), and the servers of a pool share the load: each takes the next request of any client. Each client has a ring of its own, which holds its requests and the room for their replies, so a server never waits for a client. Requests and replies are of one piece, 1 to maxMessage bytes, the owner's choice (MaxPiece()). Here the server creates the service (fipc.CreateService(name, capacity, maxMessage, maxClients, maxServers) returns the owner's *fipc.Service), joins it as its server (fipc.JoinService with fipc.Server: the owner's process admits every joiner on a thread of the library's, its own included, so one goroutine can own a service and serve) and answers three requests with their uppercase. The client joins (fipc.Client) and calls ping 1 to ping 3: Call sends a request and returns its reply. The service's Close ends it: clients' sends and servers' receives return fipc.ErrDisconnected, and a client receives the replies that came before the end, then fipc.ErrDisconnected; a client's or a server's Close never ends the service. The garbage collector never closes a service: a service lives until Close or the end of its process, and dropping the *fipc.Service doesn't end it.

package main

import (
	"bytes"
	"log"
	"time"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	// 4 KiB per client, requests and replies of up to 256 bytes, 1 client and 1 server: this process owns the service
	// and is its server
	service, err := fipc.CreateService("my_service", 4096, 256, 1, 1)
	if err != nil {
		log.Fatal(err)
	}
	server, err := fipc.JoinService("my_service", fipc.Server, 5*time.Second) // this process admits it, as any member
	if err != nil {
		log.Fatal(err)
	}
	for range 3 { // three requests, then the end
		request, err := server.Receive(fipc.Forever) // the next request of any client
		if err != nil {
			log.Fatal(err)
		}
		// the reply goes to its client, and never waits
		if err := server.Send(bytes.ToUpper(request), fipc.Forever); err != nil {
			log.Fatal(err)
		}
	}
	server.Close()
	service.Close() // ends the service: the client still receives the replies sent
}

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	client, err := fipc.JoinService("my_service", fipc.Client, 5*time.Second) // waits up to 5 s for the service
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()
	for i := 1; i <= 3; i++ {
		// sends the request and returns its reply; fipc.ErrAbandoned if the server ended before it replied
		reply, err := client.Call(fmt.Appendf(nil, "ping %d", i), fipc.Forever)
		if err != nil {
			log.Fatal(err)
		}
		fmt.Println(string(reply)) // PING 1, PING 2, PING 3
	}
}

• A client's replies come in the order of its requests. Send(request) sends a request, and a receive (Receive, ReceiveInto, or in place ReceiveAcquire until ReceiveRelease) returns the reply to its oldest request whose reply it hasn't received yet, whichever server replies first. A client may send several requests before it receives (as many as its ring has room for: each holds its length rounded up to 64 bytes, plus maxMessage rounded up to 64, until its reply has been received), and the servers work on them at once. A receive that times out or is cancelled leaves its reply to come: the next receive returns it.
• client.Call(request, timeout) never returns another request's reply. A call checks its request first: ErrInvalid for an empty one, ErrTooLarge over MaxPiece(), at once, even while a reply is owed. Then a call on a cancelled connection returns ErrCancelled at once, whatever its timeout, and sends nothing: it could never receive the reply. The connection keeps count of the replies still owed: a call whose receive timed out or was cancelled leaves its reply owed, and the next call first receives and drops that reply (at most one is owed, since every call drops it first; an ErrAbandoned in its place counts too), within its own timeout and before it sends. A call that had to drop an owed reply or wait for its turn first sends its request only if its timeout has time left: once a positive timeout has run out, it returns ErrTimeout without sending, so no server sees that request; with fipc.NoWait no step waits, and it sends if there is room. After a call that gave up, a plain Receive returns that call's late reply (or ErrAbandoned) first: replies come in request order. A call returns ErrInvalid while a request sent with Send (or SendCommit) still has its reply to be received: that reply is yours, so receive it first; on any connection other than a service's client it returns ErrInvalid too.
• Calls from several goroutines run one at a time, each returning its own request's reply. A call waits for its turn within its own timeout (ErrTimeout if another call holds the turn that long), and client.Cancel() or client.Close() ends the wait with ErrCancelled. While a call runs it holds both directions of the connection: the client's sends and receives on other goroutines wait for it to return, so that none takes the call's reply (and a call waits for one of theirs already in progress, which can outlast the call's timeout).
• A server holds one request at a time. Receive takes the next request of any client (each client's in the order it sent them), and Send(reply) replies to it and never waits; until then its receives return ErrInvalid, and a reply with no request held returns ErrInvalid. A request received in place stays valid until the reply's Send or SendCommit, so the reply may be written from it; ReceiveRelease does nothing on a server. A server may receive on one goroutine and reply on another. A reply to a client that has left is dropped.
• fipc.ErrAbandoned: a client's receive (or call) returns ErrAbandoned when the server that took the request ended before it replied: it closed its connection, or its process ended. The request counts as answered, and the next receive returns the next reply. An abandoned request may or may not have been carried out; the service never sends it again, and the client decides whether to.
• The owner's service: service.WaitForMembers(role, atLeast, timeout) returns the number of members of the role (fipc.Client, fipc.Server, or fipc.AnyServiceRole for both) joined once there are at least atLeast; service.Cancel() makes its waits return ErrCancelled. Clients wait while no server is joined, and servers while no client is.
• Not covered: a server that hangs while its process runs (there are no timeouts, by design: its client waits, within its own timeouts); the owner's process, with which the service ends.

reply, err := client.Call(request, 5*time.Second)
if errors.Is(err, fipc.ErrAbandoned) { // its server ended before it replied: send it again, if it may run twice
	reply, err = client.Call(request, 5*time.Second)
}
if err != nil {
	return nil, err // fipc.ErrDisconnected: the service ended
}

request, err := server.ReceiveAcquire(fipc.Forever) // the request in place, valid until the reply
if err != nil {
	return err
}
reply, err := server.SendAcquire(len(request), fipc.NoWait) // a reply never waits
if err != nil {
	return err
}
copy(reply, bytes.ToUpper(request))    // written from the request in place
return server.SendCommit(len(request)) // the reply; request is no longer valid

for range 4 { // a pool of four servers on goroutines: each takes the next request of any client
	go func() {
		server, err := fipc.JoinService(name, fipc.Server, 5*time.Second)
		if err != nil {
			log.Fatal(err)
		}
		defer server.Close()
		for {
			request, err := server.Receive(fipc.Forever)
			if err != nil {
				return // fipc.ErrDisconnected: the service ended
			}
			if err := server.Send(work(request), fipc.Forever); err != nil {
				return
			}
		}
	}()
}

The server.go and client.go above work with a peer in any of the ten languages: every pair of languages is tested. A Go service can serve a Python or JavaScript front end, or a game's editor in C#, without a socket.

Poll the connection once a frame with fipc.NoWait, or give the receiving to a goroutine that forwards each message through a channel the frame loop drains. In an Ebitengine game, poll in Update, which runs 60 times a second by default. game_loop.go runs against echo_server.go.

package main

import (
	"encoding/binary"
	"errors"
	"fmt"
	"log"
	"time"

	"github.com/fastipc/fastipc/bindings/go/fipc"
)

func main() {
	conn, err := fipc.ConnectStream("demo", 5*time.Second)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	echoes := 0
	buf := make([]byte, 64)
	for frame := range uint32(60) {
		update := binary.LittleEndian.AppendUint32(nil, frame)
		if err := conn.Send(update, fipc.NoWait); err != nil { // ErrTimeout only if the ring is full
			log.Fatal(err)
		}
		for {
			_, err := conn.ReceiveInto(buf, fipc.NoWait)
			if errors.Is(err, fipc.ErrTimeout) {
				break // nothing more this frame
			} else if err != nil {
				log.Fatal(err) // ErrDisconnected: the peer is gone
			}
			echoes++
		}
		time.Sleep(16 * time.Millisecond) // the rest of the frame
	}
	fmt.Printf("%d echoes in 60 frames\n", echoes)
}

messages := make(chan []byte, 64)
go func() {
	defer close(messages)
	for {
		msg, err := conn.Receive(fipc.Forever)
		if err != nil {
			return // ErrDisconnected, or ErrCancelled once the game closes the connection
		}
		messages <- msg
	}
}()
// Each frame: whatever has arrived, without waiting
for len(messages) > 0 {
	handle(<-messages)
}

Measured through this binding (Go 1.27, purego), copied 16-byte messages run at 9 to 10.5 million per second between two processes, and 64 KiB messages at 420,000 to 450,000 per second; zero-copy runs slower at 16 B, 5.5 to 7.4 million per second, since it takes two calls per message on each side (details).