Persistant connection would be nice to have

[the100rabh]

Each time PHP script gets executed, a new instance is created of Kafka. It would be nice if there was some way of keeping the connection persistent and not having to fetch meta data for each request being made. This is especially helpful in case of web services with php-fpm and nginx to avoid latencies due to meta data fetch.

Instead of creating a new topic object each time, it would reuse the same topic object across script invocations.

[bennfocus]

Is there any plan to implement this?
I think it's most important feature for PHP which running in FPM environment. Otherwise each request it have to trigger a bunch of requests to Kafka brokers which really makes it slow.

[djibomar]

@arnaud-lb Can we have this ? How can I help ?

[arnaud-lb]

I would gladly accept help on this.

Basically, we would create an API like this:

$instanceName = "somename";
if (!$rk = \RdKafka\Producer::getInstance($instanceName)) {
    $rk = new \RdKafka\Producer($conf, $instanceName);
}

\RdKafka\Producer::getInstance() would search for an existing persistent Producer instance with the given name, and return it.

\RdKafka\Producer($conf, $name) would register itself if a name is passed as second argument.

Internally, only the rdkafka handle would be persistent, not the actual PHP object. getInstance() would search the persistent rdkafka handle, and create a new Producer PHP object with the rdkafka handle.

Producer() would not destroy the rdkafka handle if it's a persistent handle.

[awons]

Hi @arnaud-lb!
Since we need the persistent producer we decided to implement something ourselves.
I already have something started (still WIP, only tested on PHP 7.1) here: https://github.com/awons/php-rdkafka/tree/issue-42/persistent-producer

I am not a C developer and this is my first time with PHP's extension so hopefully this is not too bad ;)
I managed to implement a persistent kafka instance. It works quite well (tested within our own integration tests using the extension) but still have one main issue - callbacks. When request ends all callbacks are cleared and kafka, obviously, segfaults when trying to call an arbitrary memory location.

I wanted to ask what do you think about the following solution: For persistent producers we create a hash table with all possible callbacks doing nothing. Those callbacks are then registered in kafka. Then on getInstance() we also need to provide new callbacks like getInstance('name', function(){}).
This way kafka always sees pointer to the same global callback and we only replace pointers within this callback.
After request those temporary pointers are nullified and when we forget to re-register them kafka will still call the global function doing nothing.

I am still not sure how to handle situation when different requests will need register different callbacks for the same instance, but I will try to figure something out.

What do you think about this idea?

[arnaud-lb]

@awons: I'll take a look this week!

[arnaud-lb]

Awesome!

I like the general idea.

About callbacks: You are right, we could do it that way. Maybe we could use the opaque setting to store this data (see rd_kafka_conf_set_opaque() / rd_kafka_opaque()). C would just forward calls to the user callbacks referenced in the opaque struct. We could add methods to change these callbacks after getInstance().

About the instance table: Zend hash tables can be created persistent (see zend_hash_init()).

[awons]

Totally agree with the approach to callbacks. There is just one more thing I cannot grasp. We need a way to figure out which callback needs to be called exactly. One instance can only register one callback. Hence one event can, in theory, be propagated to n different requests simultaneously. We the would need a way to register per-request callbacks somehow (maybe with request ID/thread ID?). Not sure yet how to approach this.

As for zend_hash_init. I know it can be used for persistent allocation but the issue I found is that every element in that table needs to be a zval. And php < 5 cannot store arbitrary pointers in zvals. PHP 7 can, but then we either have to keep two different implementations or drop support for php<7.
At least this is what I figured.
Maybe we could treat pointers as integers and store them like that... It is kind of a hack but maybe it could work.

[arnaud-lb]

Hence one event can, in theory, be propagated to n different requests simultaneously

Are you sharing the rdkafka instance between multiple threads ? I believe that it could become very difficult to manage. From the point of view of the user, we will receive events that are not related to the current request. From the point of view of the extension, there is a lot of work required to make sure that callbacks are called in the right thread.

It would be simpler to not share rdkafka instances between multiple threads. Sharing the instance between subsequent requests on the same thread would be enough I think (this is how it would behave on non-threaded environments, like php-fm, too).

PHP internals have a notion of per-thread globals that you can use to achieve that. This is not used in php-rdkafka currently, but you can add support for per-thread globals by applying something like this: https://gist.github.com/arnaud-lb/ed0a6501f17eee05e3f814b6ae2dccd6. Basically, the ZEND_BEGIN_MODULE_GLOBALS() declares a struct that will be thread-specific (each thread has its own instance). You can access field with the RDKAFKA_G() marco, e.g. RDKAFKA_G(some_value)=1.

As for zend_hash_init. I know it can be used for persistent allocation but the issue I found is that every element in that table needs to be a zval

Actually you can store pointers, even though it's not super obvious in PHP 5. See add_consuming_toppar() in rdkafka.c.

[awons]

Sorry for late response. Was too busy with another project.

Are you sharing the rdkafka instance between multiple threads ?.

That was my initial idea - to have this feature also available in ZTS. But right now I see it is probably way too complicated and could lead to too many issues. It is probably just not worth it.
We could simply throw a RutimeException if someone tries to use this method in ZTS mode, or not even compile it there; not sure which one would be better.

Actually you can store pointers, even though it's not super obvious in PHP 5. See add_consuming_toppar() in rdkafka.c

That is a really nifty workaround :) I will refactor instances handling to use this tick.

[awons]

I was just thinking about another thing. You mentioned module globals. Correct me if I am wrong, but if we are not going to implement this feature for ZTS then it makes no sense to use module globals because in NTS mode they always resolve to a simple variable.

[arnaud-lb]

We could simply throw a RutimeException if someone tries to use this method in ZTS mode, or not even compile it there; not sure which one would be better.

Actually there is no problem with using this feature in ZTS mode, as long as we use module globals. Module globals are the way to go when we need variables whose value persist across requests, but should not be shared across threads.

[awons]

Actually there is no problem with using this feature in ZTS mode, as long as we use module globals. Module globals are the way to go when we need variables whose value persist across requests, but should not be shared across threads.

Did my homework. Got confused by an explanation of threaded model I found somewhere on the internet. But you are 100% right. We either have one PHP interpreter per process or one per thread. One interpreter will be responsible for handling multiple requests sequentially (either withing one process or one thread), so it makes perfect sense having this feature working in ZTS mode.
Getting back to work then :)

[awons]

@arnaud-lb I think I am missing something again. You mentioned the add_consuming_toppar() method as an example of how to use raw pointers with hash tables in PHP 5. The problem I see is that methods like zend_hash_str_add_ptr and zend_hash_str_find_ptr are only available in PHP 7. PHP 5 only has zend_hash_add and zend_hash_find.

Now, I see three options here:

1. I am missing something obvious here and this should work with PHP 5 (highly doubt it);
2. This is a bug and should be fixed for the extension to work with PHP 5;
3. This is not a bug and the extension is just not meant to work with PHP 5 (but looking at all the IF/ELSE statements checking for different versions - don't really think so);

Which is it then?

[arnaud-lb]

These functions are defined in php_rdkafka_priv.h if the version of PHP doesn't have then already:

php-rdkafka/php_rdkafka_priv.h

Lines 129 to 134 in 6f6f4b0

	static inline void *zend_hash_str_add_ptr(HashTable *ht, const char *str, size_t len, void *pData)
	{
	void *pDest;
	zend_hash_add(ht, str, len, &pData, sizeof(pData), &pDest);
	return pDest;
	}