New tarantool tx thread api

[EvgenyMekhanik]

We add one data struct and two functions to public API in module.h

/** \cond public */

/* Structure describing the request to the module */
struct module_request {
	struct rlist entry;
	void (*tx_process_request)(void *);
	void *data;
};

/* Send request to tx thread from module */
API_EXPORT void
tx_msg_send(struct module_request *request);

/* Cancel module request */
API_EXPORT void
tx_msg_cancel(struct module_request *request);

/** \endcond public */

When a module wants to have done some work in tx thread, it creates struct module_request, fills it with its own function and data and passes it to tx_msg_send.

• We will use a common message queue from all modules (protected by a mutex) to tx thread and common ev_async.
• Tx thread processes all module messages in new fiber.
• If a module wants to have response from tx thread (for example select tuple), the module must implement it's own message queue (tx thread to module thread ) and implement its preferred way to get message in tx_process_request function (for this purpose we may see for thread safe queue in dpdk as i think).
• Module must use fibers in lua functions running in tx thread (if they need wait for a long time).

Lua module api and high level api described in https://hackmd.io/PUgnXG61TJGaA3uc1tXAqg

I think it would be good to add the following three functions and structure to the public api.

struct tarantool_module_vtab {
        /**
          * Pointer to the function, which called when module registered in tarantool.
          * @param[out] ctx module context which later passed to start, stop, destroy functions.
          * @param[in] module pointer to tarantool module context, which module save in its private struct
          * /
	int  (*init)(void **ctx, void *module);
        /**
          * Pointer to the function, which start main module thread
          * /
	void (*start)(void *ctx);
        /**
          * Pointer to function, which stop main module thread
          * /
	void (*stop)(void *ctx);
        /**
          * Pointer to function which destroy module private resources
          * /
	void (*destroy)(void *ctx);
};

/* Register new module in tarantool */
API_EXPORT int
tarantool_register_module(const char *name, struct tarantool_module_vtab *vtab, const char **error);

/* Start new module */
API_EXPORT int
tarantool_start_module(void *module, const char **error);

/* Stop module */
API_EXPORT int
tarantool_stop_module(void *module, const char **error, int restart);

Thus, the logic of the work looks like this. The user types module = require('module'), then luaopen_module function called. The module registers its own functions using luaL_openlib and then call tarantool_register_module, to create module context in tarantool and insert it in module list. In this function, we pass a table of pointers of the main module functions.
When user types module.cfg{} module check params and if it's ok, calls tarantool_start_module in order for tarantool to create a thread for the main module function and call inside this function start from module vtab.
When user want to stop module he types module.stop, then module calls tarantool_module_stop in order for tarantool to stop a main module thread.
I think if add this functions to public API we will be able to transfer all resource management of the module to tarantool.

If we do not do this, the module creator will have to implement quite complex work with fibers every time (for example, we can not stop module and wait it stopped in tx thread)

Here I want to add some points that we discussed at the last meeting with mons. We need to implement five function in our public module API.

/*
 * This is not public struct, it's used in tarantool to describe module
 */
struct module {
      /*
       * There will be other fields here, but there should definitely be a message queue 
       * from the tx thread to the module thread.
       */
      struct lock_free_fifo *fifo;
};

struct lock_free_fifo {
       /*
        * Next position to be written in case of multi thread writers
        */
	unsigned reserve_write;
         /*
          * Next position to be written in case of one thread writer, 
          * also last position to read
          */
	unsigned write;
        /*
         * Next position to be read, must be power of 2
         */
	unsigned read;
         /*
          * Circular buffer length
          */
	unsigned len;
        /*
         * Buffer contains pointers 
         */
	void *volatile buffer[];
};

/** \cond public */

/**
 * Structure describing module request
 */
struct request{
       /*
        * Function called in tx thread with data argument
        * /
	void (*process_request_cb)(void *);
	void *data;
};

/**
 * Try to insert msg from module thread to tx thread in tx msg queue
 * Return 0 if success, else in case queue is full return -1
 */
API_EXPORT int
tx_msg_send(struct request *request);

/**
 * Try to insert msg from tx thread to module thread in module msg queue
 * Return 0 if success, else in case queue is full return -1
 */
API_EXPORT int
thr_msg_send(void *module, struct request *request);

/**
 * Try to get num data pointers from module queue,
 * return count of pointers, which was collected (<= num)
 */
API_EXPORT unsigned int
thr_msg_collect(void *module, void **data, unsigned num);

/**
 * Register module in tarantool, return 0 if success, otherwise return -1 and save error string in *error
 */
API_EXPORT int
tarantool_register_module(void **module, const char *name, const char **error);

/**
 * Unregister module in tarantool
 */
API_EXPORT int
tarantool_unregister_module(void *module);


/** \endcond public */

I specified the function arguments as I see them now. I described the number of functions
and their purpose based on our discussion with Mons

[Mons]

Since message passing would be implemented using queue, the following rules must apply:

1. We shoud have an ability to control queues length
2. We need no tx_msg_cancel. Instead, we need response from tx_msg_send with reply: success or not.

[alyapunov]

Did I understand correctly - you propose that tx_msg_send must fail if the queue length is too high?
I don't like the idea. Queue length depends on several factors, not only load factor of TX. Another problem is that queue length is a cumulative value - only its growth (not the length itself) mean that TX if overloaded.
I think the simplest and correct method of load control is to track timeout. I a request hasn't been consumed by TX thread to some timepoint - it should be cancelled.

[Mons]

Queue length control is rule N1 of queues usage
Queue length is not cumulative. It's normal state is empty.
Timeout is suitable with request/response. For one direction messaging it's almost useless.
Take a look at different IPC or networking API's: they always have return value, which means success enqueueing to some internal queue.
Then, if you talk about timeouts, you need a way to handle them. So only one function will not be sufficient.

[Mons]

Will return later with interthread communication diagram and API proposal

[Mons]

Module lifecycle

The main entry point for module spawning and configuration is Lua.

Module should be aware of it's own threads lifetime, their startup and shutdown. Threads may be spawned or gracefully respawned inside module's mod.cfg{...} implementation. Module stop with threads stop may be implemented by hand with something like mod.stop().

Nevertheless, there is one unobvious point, that module author must be aware of: server shutdown. It is possible to implement this using on_shutdown trigger, but it is much better to provide C-level API for registering on_shutdown. So, first call we need is:

• on_shutdown(...) — It should accept function callback and an argument to that function. During server shutdown, for every such callback a fiber shold be spawned and function invoked inside it. Shutdown handler function may yield: so it is required to wait their finishing and to avoid infinite waiting a timeout for forcible shutdown should be provided (which may be counted in separate fiber).

Of cource, we need both register and unregister interface and it may be implemented either with two separate functions or like in lua API: with one function, that accepts replacement pointer:

void on_shutdown( void * , void (*handler)(void *), void (*handler)(void *));

on_shutdown( mod_ptr, my_stop_fn, NULL );
// register new handler to be called with mod_ptr

on_shutdown( mod_ptr, my_stop_fn, my_stop_fn );
// re-register same handler to be called with mod_ptr

on_shutdown( mod_ptr, my_new_stop_fn, my_stop_fn );
// re-register other handler to be called with mod_ptr

on_shutdown( mod_ptr, NULL, my_new_stop_fn );
// unregister handler my_new_stop_fn

Original

[Mons]

Cross-thread messaging API (tx -> thread)

For easy module authors experience I propose to include default and supported way for communication in direction tx → thread.

Unified and eventloop independent interface, in this case, may be the following (prefix xtm_ stands for cross-thread-messaging):

From tx side:

1. xtm_create(...) — should create/allocate an instance of api for communication with one worker thread in one direction (tx → thread). Should accept queue length limit. and have sane default one.
2. xtm_delete(...) — should cleanup resources allocated during create
3. xtm_msg_send(...) — should deliver message into queue and notify eventfd/pipe about new messages. May return success or failure (in case of queue overflow over limit)

For throughtput optimisations we may consider 2 additional separate functions:
4. xtm_msg_enqueue(...) — should enqueue message into queue without notification the pipe. (also should be aware of limit)
5. xtm_msg_notify(...) — should notify eventfd/pipe about new messages
In other words, send is enqueue+notify

From thread side:

For eventloop independent monitoring of messages we need and fd, one may watch for readability. Then, it may be used in conjunction with select(2), libev's ev_io or even with blocking loop.

6. xtm_fd() — should return fd to watch

Upon notifiction on fd we need to dequeue messages from queue. For performance and throughput reasons batch receival is preferred.

7. xtm_msg_recv() — should return a list of messages from queue

It seems, that also we need xtm_msg_free. But maybe per-message-allocation-free algorithm is available.

Diagram for this process:

Original

[Mons]

Cross-thread messaging API (thread -> tx)

As and example diagram I'd provide simple connect-accepting module.

The only required API function are

1. tx_msg_send — the function, that should dispatch a message into queue. May return success or failure (in case of queue overflow over limit)

It may be efficient to check queue availability before some step, that produce message dispatching (like connection accepting), so additional helper function will be useful

2. tx_msg_probe — shoud return true or false to indicate availability of message disatching

And referring previous API (tx->thread) one may note, that we may use the same queue and API. So we may take xtm_msg_send as tx_msg_send and we need to add xtm_msg_probe (instead of tx_msg_probe)

From the side of TX message receival and function invocation may be hidden by tarantool own implementation

Original

[EvgenyMekhanik]

https://github.com/mechanik20051988/queue ссылка на репозиторий, где содержится пример реализации соответствующих функций без привязки к тарантулу. Еще потребуется реализовать функцию tx_msg_send, для отправки сообщения в поток tx.

[EvgenyMekhanik]

Обновил репозиторий. Убрал из публичного АПИ функции xtm_msg_enqueue и xtm_msg_notify. В настоящий момент реализована следующая логика, xtm_msg_send нотифицирует читающий поток в случае если очередь на момент помещения сообщения была пуста, в противном случае сообщение просто помещается в очередь (читающий поток не успел очистить очередь, нотификация не нужна, читающий поток очистит очередь). Функции xtm_msg_enqueue и xtm_msg_notify не нужны, так как у нас используется lock free очередь (в случае с cbus нотификация требовала захвата мьютекса и поэтому мы нотифицировали читающий поток, только если накапливалось определенное количество сообщений в очереди).

[void234]

English, please. Even bad one would do fine.

(Comments below may be slightly out-of-date as repo is no longer at e52e8b657fcdddf4937188b063f783a5bcb097e4 - and force push is a bad idea).

You have two DIFFERENT definitions of "struct xtm_queue" in two .c files. Plus a declaration in header file.

You should embed "struct lock_free_fifo" into end of "struct xtm_queue", this would save memory reads in hot path.

You should use consistent naming for structs/fields/locals/files - xtm_queue/queue/lock_free_fifo/fifo are completely mixed up now.

You have 3 definitions of "struct msg". I am fine with file-local ones, but one of them is in a header file. Neither grepable nor readable.

We should probably create typedef for message content (as "void *" or more likely "uintptr_t") to make code more readable.
xtm_msg_recv() should use slightly larger buffer for "flushing" reads (we have enough stack space, right?). And read() can get EINTR. This applies to other calls to read() and write(). write() only 1 byte.

I am against "errno" usage. It is to slow (thread-local varible) and complex (two values to check instead of one). Everything should be encoded into a single return value (and documented in function declaration).

xtm_msg_notify() should never fail - it should retry in case of EINTR, "pipe is full" is not an error because it means read from other end of pipe would succeed.

You are returning either "-1" or "true" from xtm_fun_dispatch(). This is clearly wrong.

xtm_msg_recv()/xtm_fun_invoke() should not call read() at all if there is already data to process.

I do not like "n_input" approach at all. There are too many atomic operations and in-memory variables. We can use single 64-bit in-memory variable to store read and write indexes as bitfields and update it using atomic compare-and-swap. This way both reader and writer can know whether there is something in the ring already.

xtm_fun_dispatch()/xtm_fun_invoke() must not use malloc()/free() and linked lists - payload is small, max ring size is known, we can just use preallocated array and memcpy() (two of them if we are crossing end of ring buffer). I am not sure we even need intermediate buffer - right now you are not updating ring indexes until processing is done, why not use data from ring buffer directly?

[EvgenyMekhanik]

Some comments.

1. Two different definitions of xtm_queue were needed, because there were two build path in the example, in one case a queue with a ring buffer, in other case pointer queue. As a result, we decided on ring buffer, now there is no second option (fixed).
2. Files and queue names renamed (fixed).
3. Now the ring queue is template, so there can be any object, but as a result of the discussion with Mons, we stopped at the fact that there is a structure containing a pointer to the function and a pointer to the data.
4. As for errno, this is what Mons described in the task, in any case, errno is set by read/write calls, etc. and not set manually
5. Now pipe "flushing" at once (fixed).
6. We write 8 bytes, because in the future we will use eventfd for linux, (A write(2) fails with the error EINVAL if the size of the
   supplied buffer is less than 8 bytes for eventfd)
7. EINTR now it is processed correctly (fixed).
8. html_msg_notify cannot satisfy should never fail condition, because write can return an error not only when the buffer is overflowed or when a signal is sent, but also EBADF and etc.
9. xtm_fun_dispatch fixed.
10. We use read in xtm_fun_invoke in all cases because we don't now what event loop used. I offered to shift the responsibility of reading the pipe to the user (for example, for libuv, the library itself does this).
11. It's better two use two 32bit value two read and write positions, (64 bit value is bad for atomic operations in 32 bit system, also atomic_load, store faster then compare_and swap).
12. About n_input: we use this parameter to prevent unnecessary writing to the pipe (we write only if queue if was empty before we put msg). We can use
    template
    static inline unsigned
    xtm_scsp_queue_free_count(struct xtm_scsp_queue *queue)
    {
    unsigned queue_write = pm_atomic_load_explicit(&queue->write, pm_memory_order_acquire);
    unsigned queue_read = pm_atomic_load_explicit(&queue->read, pm_memory_order_acquire);
    return (queue_read - queue_write - 1) & (queue->len - 1);
    }
    But i dont think that it's faster then keep atomic counter of elements.
13. Now malloc not used, also i write function to execute msg functions without copy (fixed).

[Mons]

API Draft, v3

After an online discussion, we desided to take in action the following, layered API:

1. Opaque struct, that represent one-directional, one-reader-one-writer queue implementation with eventloop integration ability

   struct xtm_queue

2. Allocation function, used to create instance of struct xtm_queue and return its pointer

   struct xtm_queue * xtm_create(size_t size)

   returns poiner or NULL with the following possible errno:
   • ENOMEM
   • EINVAL
   • EMFILE
   • ENFILE
   • ENODEV
3. Deallocation function, used to free queue and close its internal fds

   int xtm_delete(struct xtm_queue *)

   Returns 0 on success. Otherwise -1 with errno set appropriately (as in close(2), since it implies close of fds)
   • EBADF
   • EINTR
   • EIO

Flows

Producer-to-Consumer may act with 2 different flows: message transmission or function dispatch. For every kind of flow would be created a pair of functions: xtm_msg_send/xtm_msg_recv and xtm_fun_dispatch/xtm_fun_invoke

Producer functions

4. Function for message delivery into queue and notification of internal fd (eventfd or pipe) about new messages.

   int xtm_msg_send(struct xtm_queue *queue, void *msg [, int delayed = 0 ] )

   Implementation details may imply: if queue is not empty, fd notification should be ommitted.
   Returns 0 on success. Otherwise -1 with errno set appropriately.

   • ENOBUFS — in case of run out of buffer size (queue overflow)
   • set of errors, that write(2) may return, but almost allof them should not encounter. But we should think of EAGAIN
     Non-zero optional argument delayed may be used for optimisations. delayed = 0 implies, that "xtm_msg_notify" is invoked implicitly

5. For performance optimisations xtm_msg_send may be available to split into two separate steps: message enqueue and remote notification (xtm_msg_send is reused):

   int xtm_msg_send(struct xtm_queue *queue, void *msg, int delayed = 1)

   Should only deliver message into a queue. As an implementation details it may use intermediate queue, which belongs only to producer thread to not use cross-thread synchronisation often. Returns 0 on success. Otherwise -1 with errno set appropriately. The only error is:

   • ENOBUFS — in case of run out of buffer size (queue overflow)

   int xtm_msg_notify(struct xtm_queue *queue)

   Should notify internal fd about new messages.

6. It may be useful to check, whether it possible to put message, before actual delivery try. For this reason we need

   int xtm_msg_probe(struct xtm_queue *queue)

   Returns 0 if queue has space. Otherwise -1 with errno set to ENOBUFS

7. For function dispatch flow the following function should be implemented

   int xtm_fun_dispatch(
       struct xtm_queue *queue,
       void (*fun)(void *),
       void *fun_arg,
       int delayed
   )

Consumer functions

8. For consumer, based on arbitraty event loop, the only way to get a wakeup notification — is a file descriptor.

   int xtm_fd(struct xtm_queue *queue)

   Returns file descriptor, that should be watched to become readable. When it became readable, consumer should call one of the consumer functions: xtm_msg_recv or xtm_fun_apply

9. Message receival for message transmission flow should be implemented with the following function. There are some interface alternatives (to be discussed):

   int xtm_msg_recv(struct xtm_queue *queue, void **data, int *out_data_size)

   int xtm_msg_recv(struct xtm_queue *queue, void **data)

   int xtm_msg_recv(struct xtm_queue *queue, struct msg **data)

   struct msg * xtm_msg_recv(struct xtm_queue *queue)

   Should read out excess data from pipe/eventfd to make it unreadable.

10. Message receival for function dispatch flow should be implemented with the following function. It shoud receive messages from queue and invoke the function for each message

    int xtm_fun_invoke(struct xtm_queue *queue)

    Returns non-negative value as a count of invoked functions or -1 in case of error.

Lifecycle

For tracking of lifecycle we need on_shutdown trigger registration function. (name is a subject to discuss)

void on_shutdown( void * , void (*handler)(void *), void (*handler)(void *));

// register new handler to be called with mod_ptr
on_shutdown( mod_ptr, my_stop_fn, NULL );

// re-register same handler to be called with mod_ptr
on_shutdown( mod_ptr, my_stop_fn, my_stop_fn );

// re-register other handler to be called with mod_ptr
on_shutdown( mod_ptr, my_new_stop_fn, my_stop_fn );

// unregister handler my_new_stop_fn
on_shutdown( mod_ptr, NULL, my_new_stop_fn );

[EvgenyMekhanik]

3c9bfb6 implement new module api

[EvgenyMekhanik]

In accordance with our recent discussions, I am fixing the final version of the API here.

1. we do xtm_queue_push_ptr + xtm_queue_push_fun, and the rest of the pushes we do later, if necessary, but we make sure that it will be easy to add them without changing other functions. On the receiving side, we do xtm_queue_pop_ptr and xtm_queue_invoke_funs_all. In the future, it will be possible to make xtm_queue_pop_fun without calling, if we want.
2. We make an optional descriptor for the ability to wait until the queue becomes incomplete. And we are doing it now, since this greatly affects the API of the queue, since it is part of the queue.
3. We don't touch xtm_queue_probe, because in the future, other reasons may be added why you
   can't push, and then the true/false flag may not be enough, you need errno.
4. We add xtm_queue prefix for all functions because if we add more structures to the xtm library in the
   future, they will conflict by prefix otherwise.
5. We do push and notify separately. Because if there is a flag, it will usually be passed as a constant, and it will be unclear from the code what this 0 or 1 means, and adds an additional if. All pushs will not do notify. But if in the future if we want, we will make functions like send_ptr, send_fun,..., which will immediately do notify.
6. For pop, we do getting from the queue without a flush pipe. To flush the pipe, you will need to call consume separately. Or in the future, by analogy with send, we will make recv_ptr, recv_fun. For invoke_funcs_all, we will not have a version that does consume, for it we will have to call consume by hand, if you want. Because we can't figure out how to do invoke/by analogy with send/recv.

[EvgenyMekhanik]

After last discussion add some comments here.

1. push functions sets special flag in queue internal in case when, producer fails to put to the queue, because queue was full.
2. pop/invoke functions do not notified producer thread when queue is not full. We add special function xtm_queue_was_full, to check and reset flag previously sets by producer thread.
3. I try to see goland channels for some ideas, but they used locks and special goland procedures to stop thread when queue is full/empty.