零拷贝接收和消息大小

Question

当使用零拷贝接收消息时，是否有一种方法来查询接收消息的大小？

有了这个，我(尝试)实现零拷贝：

zmq_recv(sock, buf, sizeof(buf), 0);

我也试过：

zmq_msg_t msg;
zmq_msg_init_data (&msg, buf, sizeof(buf), nullptr, NULL);
zmq_msg_recv(&msg, sock, 0);
size_t len = zmq_msg_size(&msg);

这将返回正确的大小，但不填充buf。我认为zmq_msg_init_data不适合与zmq_msg_recv一起使用，并且消息在接收上重新构建。

Answer 1

引用零副本上的指南：

没有办法在接收时执行零拷贝: ZeroMQ为您提供一个缓冲区，只要您愿意，它就可以存储，但是它不会直接将数据写入应用程序缓冲区。

零拷贝只用于发送，而不是接收。

Oh和zmq_recv返回接收到的字节数。

Answer 2

如果使用JeroMQ (纯Java for ZeroMQ)，则可以通过ZMQ.Socket.setMsgAllocator在接收端实现零拷贝。

一般来说，我们只关心大型对象的零副本，因此您可以设置一个自定义消息分配器，例如：

class MappedMemoryMsgAllocator implements MsgAllocator {

static final MsgAllocatorHeap heap = new MsgAllocatorHeap();
private final int threshold;

public MappedMemoryMsgAllocator(int threshold) {
    this.threshold = threshold;
}

public MappedMemoryMsgAllocator() {
    this(1024 * 1024 * 4);
}

@Override
public Msg allocate(int size) {
    if ((threshold > 0) && (size > threshold)) {
        try {
            return new Msg(UtilsJNI.nativeMapMemory(size, UtilsJNI.PROT_WRITE, UtilsJNI.MAP_PRIVATE));
        } catch (IOException e) {
            Log.d(TAG, "allocate:: fail to mmap", e);
            return null;
        }
    } else {
        return heap.allocate(size);
    }
}

public static void forceFreeMsgArray(zmq.Msg[] msgs) {
    if (msgs != null) {
        for (zmq.Msg msg : msgs) {
            final ByteBuffer buf = msg.buf();
            if (buf.isDirect()) {
                try {
                    UtilsJNI.nativeUnmapMemory(buf);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

本机C++代码：

extern "C"
JNIEXPORT jobject
UtilsJNI_nativeMapMemory(JNIEnv *env, jobject clazz, jlong size,
                                              jint protection, jint mapFlags) {
    void *pAddr = mmap64(NULL, size, protection, mapFlags|MAP_ANONYMOUS, -1, 0);
    if (pAddr == NULL) {
        env->ThrowNew(env->FindClass("java/io/IOException"), "fail to mmap");
        return NULL;
    } else {
        return env->NewDirectByteBuffer(pAddr, size);
    }
}

extern "C"
JNIEXPORT void
UtilsJNI_nativeUnmapMemory(JNIEnv *env, jobject clazz, jobject buffer) {
    if (munmap(env->GetDirectBufferAddress(buffer), env->GetDirectBufferCapacity(buffer)) < 0) {
        env->ThrowNew(env->FindClass("java/io/IOException"), "fail to munmap");
    }
}