Skip to content

TCP support for Machnet, when non-machnet application connect a machnet application. #54

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sarsanaee wants to merge 4 commits into
base: main
Choose a base branch
from
Open

TCP support for Machnet, when non-machnet application connect a machnet application. #54

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1c4ad22
Add TCP transport to Machnet
sarsanaee Feb 7, 2026
5f07af2
TCP working version
sarsanaee Feb 8, 2026
4837db8
Adding TCP flow test
sarsanaee Feb 9, 2026
a47ec54
Readme added on know how to do TCP exp
sarsanaee Feb 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions bindings/csharp/HelloWorld/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ static void Main(string[] args)
{
// Client
MachnetFlow_t flow = new MachnetFlow_t();
ret = MachnetShim.machnet_connect(channel_ctx, options.LocalIp, options.RemoteIp, kHelloWorldPort, ref flow);
ret = MachnetShim.machnet_connect(channel_ctx, options.LocalIp, options.RemoteIp, kHelloWorldPort, ref flow, 0);
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Replace this call with a call to managed code if possible.

Copilot uses AI. Check for mistakes.
CustomCheck(ret == 0, "machnet_connect()");

string msg = "Hello World!";
Expand All @@ -84,7 +84,7 @@ static void Main(string[] args)
else
{
Console.WriteLine("Waiting for message from client");
ret = MachnetShim.machnet_listen(channel_ctx, options.LocalIp, kHelloWorldPort);
ret = MachnetShim.machnet_listen(channel_ctx, options.LocalIp, kHelloWorldPort, 0);
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Replace this call with a call to managed code if possible.

Copilot uses AI. Check for mistakes.
CustomCheck(ret == 0, "machnet_listen()");

while (true)
Expand Down
4 changes: 2 additions & 2 deletions bindings/csharp/HelloWorld/machnet_shim.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,10 @@ public static class MachnetShim
public static extern IntPtr machnet_attach();

[DllImport(libmachnet_shim_location, CallingConvention = CallingConvention.Cdecl)]
public static extern int machnet_listen(IntPtr channel_ctx, string local_ip, UInt16 port);
public static extern int machnet_listen(IntPtr channel_ctx, string local_ip, UInt16 port, int protocol);
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minimise the use of unmanaged code.

Copilot uses AI. Check for mistakes.

[DllImport(libmachnet_shim_location, CallingConvention = CallingConvention.Cdecl)]
public static extern int machnet_connect(IntPtr channel_ctx, string local_ip, string remote_ip, UInt16 port, ref MachnetFlow_t flow);
public static extern int machnet_connect(IntPtr channel_ctx, string local_ip, string remote_ip, UInt16 port, ref MachnetFlow_t flow, int protocol);
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minimise the use of unmanaged code.

Copilot uses AI. Check for mistakes.

[DllImport(libmachnet_shim_location, CallingConvention = CallingConvention.Cdecl)]
public static extern int machnet_send(IntPtr channel_ctx, MachnetFlow_t flow, byte[] data, IntPtr dataSize);
Expand Down
8 changes: 4 additions & 4 deletions bindings/go/machnet/conversion.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,13 +45,13 @@ MachnetFlow_t __machnet_recvmsg_go(const MachnetChannelCtx_t* ctx,

int __machnet_connect_go(MachnetChannelCtx_t* ctx, uint32_t local_ip,
uint32_t remote_ip, uint16_t remote_port,
MachnetFlow_t* flow) {
return machnet_connect(ctx, local_ip, remote_ip, remote_port, flow);
MachnetFlow_t* flow, int protocol) {
return machnet_connect(ctx, local_ip, remote_ip, remote_port, flow, protocol);
}

int __machnet_listen_go(MachnetChannelCtx_t* ctx, uint32_t local_ip,
uint16_t port) {
return machnet_listen(ctx, local_ip, port);
uint16_t port, int protocol) {
return machnet_listen(ctx, local_ip, port, protocol);
}
Comment on lines 46 to 55
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Go bindings now call machnet_connect()/machnet_listen() with uint32_t IPs, but the C API expects const char* IP strings. This will compile with warnings (C) but pass invalid pointers at runtime (likely crash). Update the Go wrapper to pass string IPs (and remove ipv4_str_to_uint32 usage) or introduce separate C APIs (e.g., machnet_connect_ipv4_u32 / machnet_listen_ipv4_u32) and expose those instead.

Copilot uses AI. Check for mistakes.

MachnetFlow_t* __machnet_init_flow() {
Expand Down
8 changes: 4 additions & 4 deletions bindings/go/machnet/machnet.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,21 +69,21 @@ func Attach() *MachnetChannelCtx {
}

// Connect to the remote host and port.
func Connect(ctx *MachnetChannelCtx, local_ip string, remote_ip string, remote_port uint) (int, MachnetFlow) {
func Connect(ctx *MachnetChannelCtx, local_ip string, remote_ip string, remote_port uint, protocol int) (int, MachnetFlow) {
// Initialize the flow
var flow_ptr *C.MachnetFlow_t = C.__machnet_init_flow()

local_ip_int := ipv4_str_to_uint32(local_ip)
remote_ip_int := ipv4_str_to_uint32(remote_ip)

ret := C.__machnet_connect_go((*C.MachnetChannelCtx_t)(ctx), (C.uint)(local_ip_int), (C.uint)(remote_ip_int), C.ushort(remote_port), flow_ptr)
ret := C.__machnet_connect_go((*C.MachnetChannelCtx_t)(ctx), (C.uint)(local_ip_int), (C.uint)(remote_ip_int), C.ushort(remote_port), flow_ptr, C.int(protocol))
return (int)(ret), convert_net_flow_go(flow_ptr)
}

// Listen on the local host and port.
func Listen(ctx *MachnetChannelCtx, local_ip string, local_port uint) int {
func Listen(ctx *MachnetChannelCtx, local_ip string, local_port uint, protocol int) int {
local_ip_int := ipv4_str_to_uint32(local_ip)
ret := C.__machnet_listen_go((*C.MachnetChannelCtx_t)(ctx), (C.uint)(local_ip_int), C.ushort(local_port))
ret := C.__machnet_listen_go((*C.MachnetChannelCtx_t)(ctx), (C.uint)(local_ip_int), C.ushort(local_port), C.int(protocol))
return (int)(ret)
}

Expand Down
4 changes: 2 additions & 2 deletions bindings/go/msg_gen/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -324,14 +324,14 @@ func main() {
remote_ip, _ := jsonparser.GetString(json_bytes, "hosts_config", remote_hostname, "ipv4_addr")

// Initiate connection to the remote host.
ret, flow = machnet.Connect(channel_ctx, local_ip, remote_ip, remote_port)
ret, flow = machnet.Connect(channel_ctx, local_ip, remote_ip, remote_port, 0)
if ret != 0 {
glog.Fatal("Failed to connect to remote host.")
}
glog.Info("[CONNECTED] [", local_ip, " <-> ", remote_ip, ":", remote_port, "]")
} else {
// Listen for incoming connections.
ret = machnet.Listen(channel_ctx, local_ip, remote_port)
ret = machnet.Listen(channel_ctx, local_ip, remote_port, 0)
if ret != 0 {
glog.Fatal("Failed to listen for incoming connections.")
}
Expand Down
4 changes: 2 additions & 2 deletions bindings/js/hello_world.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ if (options.remote_ip) {
const flow = new MachnetFlow_t();
ret = machnet_shim.machnet_connect(
channel_ctx, ref.allocCString(options.local_ip),
ref.allocCString(options.remote_ip), kHelloWorldPort, flow.ref());
ref.allocCString(options.remote_ip), kHelloWorldPort, flow.ref(), 0);
customCheck(ret === 0, 'machnet_connect()');

const msg = 'Hello World!';
Expand All @@ -64,7 +64,7 @@ if (options.remote_ip) {
// Server
console.log('Waiting for message from client');
ret = machnet_shim.machnet_listen(
channel_ctx, ref.allocCString(options.local_ip), kHelloWorldPort);
channel_ctx, ref.allocCString(options.local_ip), kHelloWorldPort, 0);
customCheck(ret === 0, 'machnet_listen()');

function receive_message() {
Expand Down
4 changes: 2 additions & 2 deletions bindings/js/latency.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,11 +51,11 @@ const latencies_us = [];
const tx_flow = new MachnetFlow_t();
var ret = machnet_shim.machnet_connect(
channel_ctx, ref.allocCString(options.local_ip),
ref.allocCString(options.remote_ip), kHelloWorldPort, tx_flow.ref());
ref.allocCString(options.remote_ip), kHelloWorldPort, tx_flow.ref(), 0);
customCheck(ret === 0, 'machnet_connect()');

ret = machnet_shim.machnet_listen(
channel_ctx, ref.allocCString(options.local_ip), kHelloWorldPort);
channel_ctx, ref.allocCString(options.local_ip), kHelloWorldPort, 0);
customCheck(ret === 0, 'machnet_listen()');

if (options.is_client) {
Expand Down
4 changes: 2 additions & 2 deletions bindings/js/machnet_shim.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,8 @@ console.log('Loading libmachnet_shim');
var machnet_shim = ffi.Library(libmachnet_shim_location, {
'machnet_init': ['int', []],
'machnet_attach': ['pointer', []],
'machnet_listen': ['int', [voidPtr, charPtr, uint16]],
'machnet_connect': ['int', [voidPtr, charPtr, charPtr, uint16, MachnetFlowPtr]],
'machnet_listen': ['int', [voidPtr, charPtr, uint16, 'int']],
'machnet_connect': ['int', [voidPtr, charPtr, charPtr, uint16, MachnetFlowPtr, 'int']],
'machnet_send': ['int', [voidPtr, MachnetFlow_t, voidPtr, size_t]],
'machnet_recv': ['int', [voidPtr, voidPtr, size_t, MachnetFlowPtr]]
});
Expand Down
6 changes: 4 additions & 2 deletions bindings/js/rocksdb_client.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,14 +72,16 @@ async function machnetTransportClientAsync() {
ref.allocCString(options.local_ip),
ref.allocCString(options.remote_ip),
kRocksDbServerPort,
tx_flow.ref()
tx_flow.ref(),
0
);
customCheck(ret === 0, "machnet_connect()");

ret = machnet_shim.machnet_listen(
channel_ctx,
ref.allocCString(options.local_ip),
kRocksDbServerPort
kRocksDbServerPort,
0
);
customCheck(ret === 0, "machnet_listen()");

Expand Down
13 changes: 8 additions & 5 deletions bindings/rust/resources/machnet.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,27 +87,30 @@ void *machnet_attach();

/**
* @brief Listens for incoming messages on a specific IP and port.
* @param[in] channel The channel associated to the listener.
* @param[in] ip The local IP address to listen on.
* @param[in] channel_ctx The channel associated to the listener.
* @param[in] local_ip The local IP address to listen on.
* @param[in] port The local port to listen on.
* @param[in] protocol MACHNET_PROTO_UDP (0) or MACHNET_PROTO_TCP (1).
* @return 0 on success, -1 on failure.
*/
int machnet_listen(void *channel_ctx, const char *local_ip, uint16_t port);
int machnet_listen(void *channel_ctx, const char *local_ip, uint16_t port,
int protocol);

/**
* @brief Creates a new connection to a remote peer.
* @param[in] channel The channel associated with the connection.
* @param[in] channel_ctx The channel associated with the connection.
* @param[in] local_ip The local IP address.
* @param[in] remote_ip The remote IP address.
* @param[in] remote_port The remote port.
* @param[out] flow A pointer to a `MachnetFlow_t` structure that will be
* filled by the function upon success.
* @param[in] protocol MACHNET_PROTO_UDP (0) or MACHNET_PROTO_TCP (1).
* @return 0 on success, -1 on failure. `flow` is filled with the flow
* information on success.
*/
int machnet_connect(void *channel_ctx, const char *local_ip,
const char *remote_ip, uint16_t remote_port,
MachnetFlow_t *flow);
MachnetFlow_t *flow, int protocol);

/**
* Enqueue one message for transmission to a remote peer over the network.
Expand Down
2 changes: 2 additions & 0 deletions bindings/rust/src/bindings.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -529,6 +529,7 @@ extern "C" {
channel_ctx: *mut ::std::os::raw::c_void,
local_ip: *const ::std::os::raw::c_char,
port: u16,
protocol: ::std::os::raw::c_int,
) -> ::std::os::raw::c_int;
}
extern "C" {
Expand All @@ -539,6 +540,7 @@ extern "C" {
remote_ip: *const ::std::os::raw::c_char,
remote_port: u16,
flow: *mut MachnetFlow_t,
protocol: ::std::os::raw::c_int,
) -> ::std::os::raw::c_int;
}
extern "C" {
Expand Down
8 changes: 5 additions & 3 deletions bindings/rust/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,7 +151,7 @@ pub fn machnet_attach<'a>() -> Option<MachnetChannel<'a>> {
/// let remote_ip = "192.168.1.3";
/// let remote_port = 8080;
///
/// match machnet_connect(&mut channel, local_ip, remote_ip, remote_port) {
/// match machnet_connect(&mut channel, local_ip, remote_ip, remote_port, 0) {
/// Some(flow) => {
/// // Connection was successful, use `flow` here
/// }
Expand All @@ -166,6 +166,7 @@ pub fn machnet_connect(
local_ip: &str,
remote_ip: &str,
remote_port: u16,
protocol: i32,
) -> Option<MachnetFlow> {
unsafe {
let channel_ptr = channel.get_ptr();
Expand All @@ -180,6 +181,7 @@ pub fn machnet_connect(
remote_ip_cstr.as_ptr(),
remote_port,
flow_ptr, // &mut flow,
protocol,
);

match res {
Expand Down Expand Up @@ -226,11 +228,11 @@ pub fn machnet_connect(
/// }
/// ```
///
pub fn machnet_listen(channel: &mut MachnetChannel, local_ip: &str, local_port: u16) -> i32 {
pub fn machnet_listen(channel: &mut MachnetChannel, local_ip: &str, local_port: u16, protocol: i32) -> i32 {
unsafe {
let channel_ptr = channel.get_ptr();
let local_ip_cstr = CString::new(local_ip).unwrap();
bindings::machnet_listen(channel_ptr, local_ip_cstr.as_ptr(), local_port)
bindings::machnet_listen(channel_ptr, local_ip_cstr.as_ptr(), local_port, protocol)
Comment on lines +231 to +235
Copy link

Copilot AI Feb 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The machnet_listen() signature was updated to include a protocol argument, but the Rustdoc examples above this function (and any other docs referencing machnet_listen) still show the old 3-arg call. Update the examples to include the protocol parameter so doctests/docs don’t mislead users.

Copilot uses AI. Check for mistakes.
}
}

Expand Down
Loading
Loading