add ffi crate

Author: rochdev

benchmark/sirun/plugin-koa/internal-tracer/encoder.js

@@ -4,10 +4,8 @@ const Chunk = require('../../../../packages/dd-trace/src/encode/chunk')
 const { storage } = require('../../../../packages/datadog-core')
 const { Client } = require('./client')
 const { zeroId } = require('./id')
+const { now } = require('./now')
 
-const processStartTime = BigInt(Date.now() * 1e6)
-const processStartTicks = process.hrtime.bigint()
-const now = () => Number(processStartTime + process.hrtime.bigint() - processStartTicks)
 // const service = process.env.DD_SERVICE || 'unnamed-node-app'
 const ARRAY_OF_TWO = 0x92
 const SOFT_LIMIT = 8 * 1024 * 1024 // 8MB
@@ -171,10 +169,31 @@ class Encoder {
     if (count === 0) return
 
     const data = this.makePayload()
-    const path = `/v0.1/events`
 
     this._timer = clearTimeout(this._timer)
-    this._client.request({ data, path, count }, done)
+
+    if (process.env.WITH_NATIVE_COLLECTOR) {
+      this.flushFfi(data, done)
+    } else {
+      const path = `/v0.1/events`
+      this._client.request({ data, path, count }, done)
+    }
+  }
+
+  // TODO: Use node:ffi when it lands.
+  // https://github.com/nodejs/node/pull/46905
+  flushFfi (data, done) {
+    const path = require('path')
+    const { getNativeFunction, getBufferPointer } = require('sbffi')
+    const libPath = path.normalize(
+      path.join(__dirname, '../../../../collector/target/release/libffi.dylib')
+    )
+    const submit = getNativeFunction(libPath, 'submit', 'uint32_t', ['uint32_t', 'uint8_t *'])
+    const ptr = getBufferPointer(data)
+
+    submit(data.length, ptr)
+
+    done()
   }
 
   reset () {

benchmark/sirun/plugin-koa/internal-tracer/now.js

@@ -0,0 +1,10 @@
+'use strict'
+
+const processStartTime = BigInt(Date.now() * 1e6)
+const processStartTicks = process.hrtime.bigint()
+
+function now () {
+  Number(processStartTime + process.hrtime.bigint() - processStartTicks)
+}
+
+module.exports = { now }

collector/Cargo.lock

@@ -1,6 +1,7 @@
 [workspace]
 members = [
   "common",
+  "ffi",
   "hyper-client",
   "server"
 ]

collector/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "ffi"
+version = "0.1.0"
+edition = "2021"
+
+[lib]
+crate_type = ["cdylib"]
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+common = { path = "../common" }
+hyper-client = { path = "../hyper-client" }
+hyper = "0.14.24"
+tokio = "1.25.0"
+libc = "0.2.139"

collector/ffi/Cargo.toml

@@ -0,0 +1,34 @@
+use common::exporting::agent::AgentExporter;
+use common::processing::Processor;
+use hyper::body::Buf;
+use hyper_client::HyperClient;
+use tokio::sync::mpsc::{self, Receiver, Sender};
+
+extern crate libc;
+
+#[no_mangle]
+pub extern "C" fn submit(size: usize, ptr: *const u8) -> u32 {
+    internal_submit(unsafe {
+        std::slice::from_raw_parts(ptr as *const u8, size as usize)
+    }) as u32
+}
+
+#[tokio::main]
+async fn internal_submit(payload: &[u8]) -> u32 {
+    let (tx, mut rx): (Sender<()>, Receiver<()>) = mpsc::channel(1);
+
+    let mut client = Box::new(HyperClient::new());
+
+    client.on_response(tx);
+
+    let exporter = Box::new(AgentExporter::new(client));
+    let mut processor = Processor::new(exporter);
+    let mut rd = payload.reader();
+
+    processor.process(&mut rd);
+    processor.flush();
+
+    rx.recv().await.unwrap();
+
+    0 // TODO: Return proper response buffer instead.
+}

collector/ffi/src/lib.rs

@@ -1,10 +1,20 @@
 use common::client::Client;
+use tokio::sync::mpsc::Sender;
 
-pub struct HyperClient {}
+pub struct HyperClient {
+    tx: Option<Sender<()>>
+}
 
 impl HyperClient {
     pub fn new() -> Self {
-        Self {}
+        Self {
+            tx: None
+        }
+    }
+
+    // TODO: Require a sender in `new()` instead.
+    pub fn on_response (&mut self, tx: Sender<()>) {
+        self.tx = Some(tx);
     }
 }
 
@@ -29,8 +39,17 @@ impl Client for HyperClient {
             .body(hyper::Body::from(data))
             .unwrap();
 
+        let tx = self.tx.clone();
+
         tokio::spawn(async move {
-            hyper::Client::new().request(req).await.unwrap();
+            let res = hyper::Client::new().request(req).await.unwrap();
+
+            // Discard the response for now.
+            hyper::body::to_bytes(res.into_body()).await.unwrap();
+
+            if let Some(tx) = tx {
+                tx.send(()).await.unwrap();
+            }
         });
     }
 }

collector/hyper-client/src/lib.rs

@@ -86,6 +86,7 @@
     "path-to-regexp": "^0.1.2",
     "protobufjs": "^7.1.2",
     "retry": "^0.10.1",
+    "sbffi": "bengl/sbffi",
     "semver": "^5.5.0"
   },
   "devDependencies": {

package.json

@@ -2646,6 +2646,11 @@ node-abort-controller@^3.0.1:
   resolved "https://registry.yarnpkg.com/node-abort-controller/-/node-abort-controller-3.0.1.tgz#f91fa50b1dee3f909afabb7e261b1e1d6b0cb74e"
   integrity sha512-/ujIVxthRs+7q6hsdjHMaj8hRG9NuWmwrz+JdRwZ14jdFoKSkm+vDsCbF9PLpnSqjaWQJuTmVtcWHNLr+vrOFw==
 
+node-addon-api@^6.0.0:
+  version "6.0.0"
+  resolved "https://registry.yarnpkg.com/node-addon-api/-/node-addon-api-6.0.0.tgz#cfb3574e6df708ff71a30db6c4762d9e06e11c27"
+  integrity sha512-GyHvgPvUXBvAkXa0YvYnhilSB1A+FRYMpIVggKzPZqdaZfevZOuzfWzyvgzOwRLHBeo/MMswmJFsrNF4Nw1pmA==
+
 node-gyp-build@^3.9.0:
   version "3.9.0"
   resolved "https://registry.yarnpkg.com/node-gyp-build/-/node-gyp-build-3.9.0.tgz#53a350187dd4d5276750da21605d1cb681d09e25"
@@ -3190,6 +3195,12 @@ safe-regex-test@^1.0.0:
   resolved "https://registry.yarnpkg.com/safer-buffer/-/safer-buffer-2.1.2.tgz#44fa161b0187b9549dd84bb91802f9bd8385cd6a"
   integrity sha512-YZo3K82SD7Riyi0E1EQPojLz7kpepnSQI9IyPbHHg1XXXevb5dJI7tpyN2ADxGcQbHG7vcyRHk0cbwqcQriUtg==
 
+sbffi@bengl/sbffi:
+  version "1.0.4"
+  resolved "https://codeload.github.com/bengl/sbffi/tar.gz/35a2584cec76060b7a883f0ece5c956e92d01144"
+  dependencies:
+    node-addon-api "^6.0.0"
+
 semver@5.3.0:
   version "5.3.0"
   resolved "https://registry.yarnpkg.com/semver/-/semver-5.3.0.tgz#9b2ce5d3de02d17c6012ad326aa6b4d0cf54f94f"