120 lines
2.9 KiB
C++
120 lines
2.9 KiB
C++
#include <chrono>
|
|
#include <cassert>
|
|
#include <iostream>
|
|
#include <thread>
|
|
|
|
#include "ipc.h"
|
|
#include "visionipc_client.h"
|
|
#include "visionipc_server.h"
|
|
|
|
VisionIpcClient::VisionIpcClient(std::string name, VisionStreamType type, bool conflate, cl_device_id device_id, cl_context ctx) : name(name), type(type), device_id(device_id), ctx(ctx) {
|
|
msg_ctx = Context::create();
|
|
sock = SubSocket::create(msg_ctx, get_endpoint_name(name, type), "127.0.0.1", conflate, false);
|
|
|
|
poller = Poller::create();
|
|
poller->registerSocket(sock);
|
|
}
|
|
|
|
// Connect is not thread safe. Do not use the buffers while calling connect
|
|
bool VisionIpcClient::connect(bool blocking){
|
|
connected = false;
|
|
|
|
// Cleanup old buffers on reconnect
|
|
for (size_t i = 0; i < num_buffers; i++){
|
|
buffers[i].free();
|
|
}
|
|
num_buffers = 0;
|
|
|
|
// Connect to server socket and ask for all FDs of type
|
|
std::string path = "/tmp/visionipc_" + name;
|
|
|
|
int socket_fd = -1;
|
|
while (socket_fd < 0) {
|
|
socket_fd = ipc_connect(path.c_str());
|
|
|
|
if (socket_fd < 0) {
|
|
if (blocking){
|
|
std::cout << "VisionIpcClient connecting" << std::endl;
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(100));
|
|
} else {
|
|
return false;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Send stream type to server to request FDs
|
|
int r = ipc_sendrecv_with_fds(true, socket_fd, &type, sizeof(type), nullptr, 0, nullptr);
|
|
assert(r == sizeof(type));
|
|
|
|
// Get FDs
|
|
int fds[VISIONIPC_MAX_FDS];
|
|
VisionBuf bufs[VISIONIPC_MAX_FDS];
|
|
r = ipc_sendrecv_with_fds(false, socket_fd, &bufs, sizeof(bufs), fds, VISIONIPC_MAX_FDS, &num_buffers);
|
|
|
|
assert(num_buffers > 0);
|
|
assert(r == sizeof(VisionBuf) * num_buffers);
|
|
|
|
// Import buffers
|
|
for (size_t i = 0; i < num_buffers; i++){
|
|
buffers[i] = bufs[i];
|
|
buffers[i].fd = fds[i];
|
|
buffers[i].import();
|
|
if (buffers[i].rgb) {
|
|
buffers[i].init_rgb(buffers[i].width, buffers[i].height, buffers[i].stride);
|
|
} else {
|
|
buffers[i].init_yuv(buffers[i].width, buffers[i].height);
|
|
}
|
|
|
|
if (device_id) buffers[i].init_cl(device_id, ctx);
|
|
}
|
|
|
|
connected = true;
|
|
return true;
|
|
}
|
|
|
|
VisionBuf * VisionIpcClient::recv(VisionIpcBufExtra * extra, const int timeout_ms){
|
|
auto p = poller->poll(timeout_ms);
|
|
|
|
if (!p.size()){
|
|
return nullptr;
|
|
}
|
|
|
|
Message * r = sock->receive(true);
|
|
if (r == nullptr){
|
|
return nullptr;
|
|
}
|
|
|
|
// Get buffer
|
|
assert(r->getSize() == sizeof(VisionIpcPacket));
|
|
VisionIpcPacket *packet = (VisionIpcPacket*)r->getData();
|
|
|
|
assert(packet->idx < num_buffers);
|
|
VisionBuf * buf = &buffers[packet->idx];
|
|
|
|
if (buf->server_id != packet->server_id){
|
|
connected = false;
|
|
delete r;
|
|
return nullptr;
|
|
}
|
|
|
|
if (extra) {
|
|
*extra = packet->extra;
|
|
}
|
|
|
|
buf->sync(VISIONBUF_SYNC_TO_DEVICE);
|
|
delete r;
|
|
return buf;
|
|
}
|
|
|
|
|
|
|
|
VisionIpcClient::~VisionIpcClient(){
|
|
for (size_t i = 0; i < num_buffers; i++){
|
|
buffers[i].free();
|
|
}
|
|
|
|
delete sock;
|
|
delete poller;
|
|
delete msg_ctx;
|
|
}
|