json - 如何将 float 组(无需序列化/反序列化)从 Scala (JeroMQ) 传输到 C (ZMQ)？

Question

目前，我正在使用 JSON 库在发送方 (JeroMQ) 处序列化数据，并在接收方 (C、ZMQ) 处反序列化。但是，在解析时，JSON 库开始消耗大量内存，操作系统会终止该进程。因此，我想按原样发送 float 组，即不使用 JSON。

现有的发送器代码如下(syn0 和 syn1 是 Double 数组)。如果 syn0 和 syn1 各约为 100 MB，则在解析接收到的数组时会终止该进程，即下面代码片段的最后一行:

import org.zeromq.ZMQ
import com.codahale.jerkson
socket.connect("tcp://localhost:5556")

socket.send(json.JSONObject(Map("syn0"->json.JSONArray(List.fromArray(syn0Global)))).toString())
println("SYN0 Request sent”)
val reply_syn0 = socket.recv(0)
println("Response received after syn0: " + new String(reply_syn0))
logInfo("Sending Syn1 request … , size : " + syn1Global.length )

socket.send(json.JSONObject(Map("syn1"->json.JSONArray(List.fromArray(syn1Global)))).toString())
println("SYN1 Request sent")
val reply_syn1 = socket.recv(0)

socket.send(json.JSONObject(Map("foldComplete"->"Done")).toString())
println("foldComplete sent")
//  Get the reply.
val reply_foldComplete = socket.recv(0)
val processedSynValuesJson = new String(reply_foldComplete)
val processedSynValues_jerkson =   jerkson.Json.parse[Map[String,List[Double]]](processedSynValuesJson)

可以不使用 JSON 来传输这些数组吗？

这里我在两个 C 程序之间传输 float 组:

//client.c
int main (void)
{
printf ("Connecting to hello world server…\n");
void *context = zmq_ctx_new ();
void *requester = zmq_socket (context, ZMQ_REQ);
zmq_connect (requester, "tcp://localhost:5555");

int request_nbr;
float send_buffer[10];
float recv_buffer[10];

for(int i = 0; i < 10; i++)
    send_buffer[i] = i;

for (request_nbr = 0; request_nbr != 10; request_nbr++) {
    //char buffer [10];
    printf ("Sending Hello %d…\n", request_nbr);
    zmq_send (requester, send_buffer, 10*sizeof(float), 0);
    zmq_recv (requester, recv_buffer, 10*sizeof(float), 0);
    printf ("Received World %.3f\n", recv_buffer[5]);
}
zmq_close (requester);
zmq_ctx_destroy (context);
return 0;
}

//server.c

int main (void)
{
//  Socket to talk to clients
void *context = zmq_ctx_new ();
void *responder = zmq_socket (context, ZMQ_REP);
int rc = zmq_bind (responder, "tcp://*:5555");
assert (rc == 0);
float recv_buffer[10];
float send_buffer[10];
while (1) {
    //char buffer [10];
    zmq_recv (responder, recv_buffer, 10*sizeof(float), 0);
    printf ("Received Hello\n");
    for(int i = 0; i < 10; i++)
            send_buffer[i] = recv_buffer[i]+5;
    zmq_send (responder, send_buffer, 10*sizeof(float), 0);
}
return 0;
}

最后，我尝试使用 Scala 做类似的事情，但没有成功(下面是客户端代码):

def main(args: Array[String]) {
val context = ZMQ.context(1)
val socket = context.socket(ZMQ.REQ)

println("Connecting to hello world server…")
socket.connect ("tcp://localhost:5555")
val msg : Array[Float] = Array(1,2,3,4,5,6,7,8,9,10)
val bbuf = java.nio.ByteBuffer.allocate(4*msg.length)
bbuf.asFloatBuffer.put(java.nio.FloatBuffer.wrap(msg))


for (request_nbr <- 1 to 10)  {
    socket.sendByteBuffer(bbuf,0)

}
}

Answer 1

SER/DES ？大小？
不，与传输理念相关的潜在约束很重要。

您已开始使用 0.1 GB 传输有效负载大小，并报告了 JSON 库分配，以使您的操作系统终止进程。

接下来，在其他帖子中，您请求了 0.762 GB 传输有效负载的大小。

但是，ZeroMQ 传输编排中有一个比选择外部数据序列化器 SER/DES 策略更重要的问题。

没有人会禁止您尝试发送尽可能大的 BLOB，而 JSON 修饰的字符串已经向您展示了尽管存在这种方法的阴暗面，但还有其他原因不继续这样做。

ZeroMQ 毫无疑问是一个伟大而强大的工具箱。不过，人们仍然需要一些时间才能获得真正智能且高性能的代码部署所必需的洞察力，从而最大限度地发挥这一强大的主力。

功能丰富的内部生态系统“幕后”的副作用之一是隐藏在消息传递概念中的不太为人所知的策略。

可以发送任何合理大小的消息，但不保证一定能送达。它要么完全交付，或者什么都没有完全交付，如上所述，没有任何保证。

哎呀？!

是的，但不能保证。

基于这一核心的零保证理念，在决定步骤和措施时应谨慎行事，如果您打算尝试将Gigabyte BEAST移至那里，则更应谨慎然后返回。

从这个意义上说，它可能会得到真正的SUT测试的定量支持，小尺寸的消息可以传输(如果您确实仍然需要移动GB(请参阅上面的评论，在OP下)并且没有其他选择)将整个数据量分割成更小的部分，并采用容易出错的重新组装措施，这会导致更快、更安全最终-到最终的解决方案，而不是尝试使用哑力并指示代码将大约 GB 的数据转储到实际可用的任何资源上(ZeroMQ 的零复制原理本身不能也不会节省您的这些努力)。

有关另一个隐藏陷阱的详细信息，与不完全零复制实现相关，read Martin SUSTRIK's, co-father of ZeroMQ, remarks on Zero-Copy "till-kernel-boundary-only" (因此，预计内存空间分配至少是两倍......)。

<小时/>

解决方案:

重新设计架构，以便传播小尺寸消息，如果不在远程进程中保留原始数据结构“镜像”，而不是尝试保持一次性千兆传输的生存能力。

<小时/>

下一步最好的步骤是什么？

虽然几个 SLOC 并不能解决你的问题，但如果你真的想将你的智力投入到分布式处理中，最好的办法是阅读 Pieter HINTJEN 的可爱书籍“Code Connected，Vol.1”

是的，产生自己的见解需要一些时间，但这将在许多方面将您提升到专业代码设计的另一个水平。 值得花时间。值得努力。