#include "json.h"
#include "list.h"
#include "ofpbuf.h"
+#include "ovs-thread.h"
#include "poll-loop.h"
#include "reconnect.h"
#include "stream.h"
/* Input. */
struct byteq input;
+ uint8_t input_buffer[512];
struct json_parser *parser;
struct jsonrpc_msg *received;
rpc = xzalloc(sizeof *rpc);
rpc->name = xstrdup(stream_get_name(stream));
rpc->stream = stream;
- byteq_init(&rpc->input);
+ byteq_init(&rpc->input, rpc->input_buffer, sizeof rpc->input_buffer);
list_init(&rpc->output);
return rpc;
jsonrpc_received(rpc);
if (rpc->status) {
const struct byteq *q = &rpc->input;
- if (q->head <= BYTEQ_SIZE) {
+ if (q->head <= q->size) {
stream_report_content(q->buffer, q->head,
STREAM_JSONRPC,
THIS_MODULE, rpc->name);
jsonrpc_recv_wait(struct jsonrpc *rpc)
{
if (rpc->status || rpc->received || !byteq_is_empty(&rpc->input)) {
- (poll_immediate_wake)(rpc->name);
+ poll_immediate_wake_at(rpc->name);
} else {
stream_recv_wait(rpc->stream);
}
static struct json *
jsonrpc_create_id(void)
{
- static unsigned int id;
- return json_integer_create(id++);
+ static atomic_uint next_id = ATOMIC_VAR_INIT(0);
+ unsigned int id;
+
+ atomic_add(&next_id, 1, &id);
+ return json_integer_create(id);
}
struct jsonrpc_msg *