encoderd: support remote encoder behind env var (#24490)

* remote encoder

* remote encoder on PC, fix type

Co-authored-by: Comma Device <device@comma.ai>

selfdrive/loggerd/loggerd.cc

@@ -1,6 +1,6 @@
 #include "selfdrive/loggerd/loggerd.h"
 #include "selfdrive/loggerd/remote_encoder.h"
-bool USE_REMOTE_ENCODER = false;
+bool env_remote_encoder = getenv("REMOTE_ENCODER") != NULL;
 
 ExitHandler do_exit;
 
@@ -202,7 +202,7 @@ void loggerd_thread() {
 
   // subscribe to all socks
   for (const auto& it : services) {
-    const bool encoder = USE_REMOTE_ENCODER & (strcmp(it.name+strlen(it.name)-strlen("EncodeData"), "EncodeData") == 0);
+    const bool encoder = env_remote_encoder && (strcmp(it.name+strlen(it.name)-strlen("EncodeData"), "EncodeData") == 0);
     if (!it.should_log && !encoder) continue;
     LOGD("logging %s (on port %d)", it.name, it.port);
 
@@ -228,7 +228,11 @@ void loggerd_thread() {
   std::vector<std::thread> encoder_threads;
   for (const auto &cam : cameras_logged) {
     if (cam.enable) {
-      encoder_threads.push_back(std::thread(encoder_thread, &s, cam));
+      if (env_remote_encoder) {
+        if (cam.has_qcamera) { s.max_waiting++; }
+      } else {
+        encoder_threads.push_back(std::thread(encoder_thread, &s, cam));
+      }
       if (cam.trigger_rotate) s.max_waiting++;
     }
   }

selfdrive/loggerd/remote_encoder.cc

@@ -7,11 +7,17 @@ int handle_encoder_msg(LoggerdState *s, Message *msg, std::string &name, struct
     ((name == "qRoadEncodeData") ? qcam_info : cameras_logged[0]));
   if (!cam_info.record) return 0; // TODO: handle this by not subscribing
 
+  // rotation happened, process the queue (happens before the current message)
   int bytes_count = 0;
+  if (re.logger_segment != s->rotate_segment) {
+    re.logger_segment = s->rotate_segment;
+    for (auto &qmsg: re.q) {
+      bytes_count += handle_encoder_msg(s, qmsg, name, re);
+    }
+    re.q.clear();
+  }
 
-  // TODO: AlignedBuffer is making a copy and allocing
-  //AlignedBuffer aligned_buf;
-  //capnp::FlatArrayMessageReader cmsg(aligned_buf.align(msg->getData(), msg->getSize()));
+  // extract the message
   capnp::FlatArrayMessageReader cmsg(kj::ArrayPtr<capnp::word>((capnp::word *)msg->getData(), msg->getSize()));
   auto event = cmsg.getRoot<cereal::Event>();
   auto edata = (name == "driverEncodeData") ? event.getDriverEncodeData() :
@@ -20,15 +26,6 @@ int handle_encoder_msg(LoggerdState *s, Message *msg, std::string &name, struct
   auto idx = edata.getIdx();
   auto flags = idx.getFlags();
 
-  // rotation happened, process the queue (happens before the current message)
-  if (re.logger_segment != s->rotate_segment) {
-    re.logger_segment = s->rotate_segment;
-    for (auto &qmsg: re.q) {
-      bytes_count += handle_encoder_msg(s, qmsg, name, re);
-    }
-    re.q.clear();
-  }
-
   if (!re.writer) {
     // only create on iframe
     if (flags & V4L2_BUF_FLAG_KEYFRAME) {
@@ -38,9 +35,8 @@ int handle_encoder_msg(LoggerdState *s, Message *msg, std::string &name, struct
         re.dropped_frames = 0;
       }
       re.writer.reset(new VideoWriter(s->segment_path,
-        cam_info.filename, !cam_info.is_h265,
-        cam_info.frame_width, cam_info.frame_height,
-        cam_info.fps, cam_info.is_h265 ? cereal::EncodeIndex::Type::FULL_H_E_V_C : cereal::EncodeIndex::Type::QCAMERA_H264));
+        cam_info.filename, idx.getType() != cereal::EncodeIndex::Type::FULL_H_E_V_C,
+        cam_info.frame_width, cam_info.frame_height, cam_info.fps, idx.getType()));
       // write the header
       auto header = edata.getHeader();
       re.writer->write((uint8_t *)header.begin(), header.size(), idx.getTimestampEof()/1000, true, false);
@@ -74,8 +70,10 @@ int handle_encoder_msg(LoggerdState *s, Message *msg, std::string &name, struct
     if (name == "roadEncodeData") { evt.setRoadEncodeIdx(idx); }
     auto new_msg = bmsg.toBytes();
     logger_log(&s->logger, (uint8_t *)new_msg.begin(), new_msg.size(), true);   // always in qlog?
-    delete msg;
     bytes_count += new_msg.size();
+
+    // this frees the message
+    delete msg;
   }
 
   return bytes_count;