tracing

Author: Alexey Andreev

Cargo.toml

@@ -14,6 +14,12 @@ libc = "0.2"
 crossbeam-utils = "0.8"
 serde = "1.0"
 crossbeam-channel = "0.5"
+opentelemetry = { version = "0.17", features = ["rt-tokio"] }
+opentelemetry-zipkin = { version = "0.15", features = ["reqwest-client"], default-features = false }
+opentelemetry-jaeger = { version = "0.16.0", features = ["rt-tokio", "reqwest_collector_client"] }
+tracing-opentelemetry = "0.17.3"
+tracing = "0.1.35"
+tracing-subscriber = "0.3.11"
 
 [dependencies.mlua]
 version = "0.6"

src/fiber_pool.rs

@@ -3,8 +3,10 @@ use std::{collections::LinkedList, rc::Rc, time::Duration};
 use crossbeam_channel::{unbounded, TryRecvError};
 use mlua::Lua;
 use tarantool::fiber;
+use tracing_opentelemetry::OpenTelemetrySpanExt;
+// use tracing;
 
-use crate::{ChannelError, Executor, ModuleConfig, Task};
+use crate::{ChannelError, Executor, ModuleConfig, Task, InstrumentedTask};
 
 struct SchedulerArgs<'a> {
     lua: &'a Lua,
@@ -26,7 +28,7 @@ fn scheduler_f(args: Box<SchedulerArgs>) -> i32 {
     } = *args;
 
     let cond = Rc::new(fiber::Cond::new());
-    let (tx, rx) = unbounded::<Task>();
+    let (tx, rx) = unbounded::<InstrumentedTask>();
 
     let mut workers = LinkedList::new();
     for _ in 0..fibers {
@@ -71,7 +73,7 @@ fn scheduler_f(args: Box<SchedulerArgs>) -> i32 {
 struct WorkerArgs<'a> {
     cond: Rc<fiber::Cond>,
     lua: &'a Lua,
-    rx: crossbeam_channel::Receiver<Task>,
+    rx: crossbeam_channel::Receiver<InstrumentedTask>,
     fiber_standby_timeout: f64,
 }
 fn worker_f(args: Box<WorkerArgs>) -> i32 {
@@ -87,7 +89,13 @@ fn worker_f(args: Box<WorkerArgs>) -> i32 {
         .create_function(move |lua, _: ()| {
             loop {
                 match rx.try_recv() {
-                    Ok(task) => match task(lua) {
+                    Ok((func, span_ctx)) => match {
+                        let span = tracing::span!(tracing::Level::TRACE, "fiber pool: exec");
+                        span.set_parent(span_ctx);
+                        let _ = span.enter();
+
+                        func(lua, span.context())
+                    } {
                         Ok(()) => (),
                         Err(ChannelError::TXChannelClosed) => continue,
                         Err(err) => break Err(mlua::Error::external(err)),

src/txapi.rs

@@ -1,15 +1,19 @@
-use tokio::sync::oneshot;
+use crate::eventfd;
 use async_channel;
 use async_channel::TryRecvError;
-use thiserror::Error;
-use crate::eventfd;
-use std::os::unix::io::{AsRawFd, RawFd};
-use std::io;
 use mlua::Lua;
+use tracing_opentelemetry::OpenTelemetrySpanExt;
+use std::io;
+use std::os::unix::io::{AsRawFd, RawFd};
+use thiserror::Error;
+use tokio::sync::oneshot;
+use tracing::Instrument;
+use opentelemetry::Context;
 
-pub type Task = Box<dyn FnOnce(&Lua) -> Result<(), ChannelError> + Send>;
-type TaskSender = async_channel::Sender<Task>;
-type TaskReceiver = async_channel::Receiver<Task>;
+pub type Task = Box<dyn FnOnce(&Lua, Context) -> Result<(), ChannelError> + Send>;
+pub type InstrumentedTask = (Task, Context);
+type TaskSender = async_channel::Sender<InstrumentedTask>;
+type TaskReceiver = async_channel::Receiver<InstrumentedTask>;
 
 #[derive(Error, Debug)]
 pub enum ChannelError {
@@ -40,24 +44,33 @@ impl Dispatcher {
         })
     }
 
+    #[tracing::instrument(level = "trace", skip_all)]
     pub async fn call<Func, Ret>(&self, func: Func) -> Result<Ret, ChannelError>
-        where
-            Ret: Send + 'static,
-            Func: FnOnce(&Lua) -> Ret,
-            Func: Send + 'static,
+    where
+        Ret: Send + 'static,
+        Func: FnOnce(&Lua, Context) -> Ret,
+        Func: Send + 'static,
     {
+        tracing::event!(tracing::Level::TRACE, "bass drop begin");
+
         let (result_tx, result_rx) = oneshot::channel();
-        let handler_func: Task = Box::new(move |lua| {
+        let result_rx = result_rx
+            .instrument(tracing::span!(tracing::Level::TRACE, "result_rx"));
+        let result_rx_span_ctx = result_rx.span().context();
+
+        let handler_func: Task = Box::new(move |lua, exec_ctx| {
             if result_tx.is_closed() {
-                return Err(ChannelError::TXChannelClosed)
+                return Err(ChannelError::TXChannelClosed);
             };
 
-            let result = func(lua);
-            result_tx.send(result).or(Err(ChannelError::TXChannelClosed))
+            let result = func(lua, exec_ctx);
+            result_tx
+                .send(result)
+                .or(Err(ChannelError::TXChannelClosed))
         });
-        
+
         let task_tx_len = self.task_tx.len();
-        if let Err(_channel_closed) = self.task_tx.send(handler_func).await {
+        if let Err(_channel_closed) = self.task_tx.send((handler_func, result_rx_span_ctx)).await {
             return Err(ChannelError::TXChannelClosed);
         }
 
@@ -67,15 +80,18 @@ impl Dispatcher {
             }
         }
 
-        result_rx.await.or(Err(ChannelError::RXChannelClosed))
+        tracing::event!(tracing::Level::TRACE, "bass drop end");
+
+        result_rx
+            .await
+            .or(Err(ChannelError::RXChannelClosed))
     }
 
     pub fn len(&self) -> usize {
         self.task_tx.len()
     }
 }
 
-
 pub struct Executor {
     task_rx: TaskReceiver,
     eventfd: eventfd::EventFd,
@@ -86,10 +102,23 @@ impl Executor {
         Self { task_rx, eventfd }
     }
 
+    // #[tracing::instrument(level = "trace", skip_all)]
     pub fn exec(&self, lua: &Lua, coio_timeout: f64) -> Result<(), ChannelError> {
         loop {
-            match self.task_rx.try_recv() {
-                Ok(func) => return func(lua),
+            // tracing::event!(tracing::Level::TRACE, "exec: iteration");
+
+            // let _ = tracing::trace_span!("executing task").enter();
+            match self.task_rx.try_recv()
+             {
+                Ok((func, span_ctx)) => {
+                    // tracing::event!(tracing::Level::TRACE, "task: start");
+                    println!("{:?}", span_ctx);
+
+                    let res = func(lua, span_ctx);
+
+                    // tracing::event!(tracing::Level::TRACE, "task: finish");
+                    return res;
+                }
                 Err(TryRecvError::Empty) => (),
                 Err(TryRecvError::Closed) => return Err(ChannelError::RXChannelClosed),
             };
@@ -98,7 +127,8 @@ impl Executor {
         }
     }
 
-    pub fn pop_many(&self, max_tasks: usize, coio_timeout: f64) -> Result<Vec<Task>, ChannelError> {
+    // #[tracing::instrument(level = "trace", skip_all)]
+    pub fn pop_many(&self, max_tasks: usize, coio_timeout: f64) -> Result<Vec<InstrumentedTask>, ChannelError> {
         if self.task_rx.is_empty() {
             let _ = self.eventfd.coio_read(coio_timeout);
         }
@@ -137,5 +167,8 @@ pub fn channel(buffer: usize) -> io::Result<(Dispatcher, Executor)> {
     let (task_tx, task_rx) = async_channel::bounded(buffer);
     let efd = eventfd::EventFd::new(0, false)?;
 
-    Ok((Dispatcher::new(task_tx, efd.try_clone()?), Executor::new(task_rx, efd)))
+    Ok((
+        Dispatcher::new(task_tx, efd.try_clone()?),
+        Executor::new(task_rx, efd),
+    ))
 }