encounter the problem with libdispatch

have to roll out my own version of dispatch_main_queue :-(

Author: liuliu

lib/ccv_sift.c

@@ -54,7 +54,7 @@ inline static double _ccv_keypoint_interpolate(float N9[3][9], int ix, int iy, i
 	{
 		double maxa = 0;
 		double maxabsa = 0;
-		int maxi = -1;
+		int maxi = j;
 		double tmp;
 
 		/* look for the maximally stable pivot */

serve/async.c

@@ -0,0 +1,61 @@
+#include <stdlib.h>
+#include <ev.h>
+#include <dispatch/dispatch.h>
+
+typedef struct {
+	void *context;
+	void (*cb)(void*);
+} main_async_t;
+
+static dispatch_semaphore_t async_queue_semaphore;
+static int queue_position = 0;
+static int queue_pending = 0;
+static int queue_length = 10;
+static main_async_t* async_queue;
+static ev_async main_async;
+
+void dispatch_main_async_f(void* context, void (*cb)(void*))
+{
+	dispatch_semaphore_wait(async_queue_semaphore, DISPATCH_TIME_FOREVER);
+	++queue_pending;
+	if (queue_pending > queue_length)
+	{
+		queue_length = (queue_length * 3 + 1) / 2;
+		async_queue = (main_async_t*)realloc(async_queue, sizeof(main_async_t) * queue_length);
+	}
+	async_queue[queue_position].context = context;
+	async_queue[queue_position].cb = cb;
+	queue_position = (queue_position + 1) % queue_length;
+	dispatch_semaphore_signal(async_queue_semaphore);
+	ev_async_send(EV_DEFAULT_ &main_async);
+}
+
+static void main_async_dispatch(EV_P_ ev_async* w, int revents)
+{
+	dispatch_semaphore_wait(async_queue_semaphore, DISPATCH_TIME_FOREVER);
+	while (queue_pending > 0)
+	{
+		queue_position = (queue_position + queue_length - 1) % queue_length;
+		--queue_pending;
+		async_queue[queue_position].cb(async_queue[queue_position].context);
+	}
+	dispatch_semaphore_signal(async_queue_semaphore);
+}
+
+void main_async_init(void)
+{
+	async_queue_semaphore = dispatch_semaphore_create(1);
+	async_queue = (main_async_t*)malloc(sizeof(main_async_t) * queue_length);
+	ev_async_init(&main_async, main_async_dispatch);
+}
+
+void main_async_start(EV_P)
+{
+	ev_async_start(EV_A_ &main_async);
+}
+
+void main_async_destroy(void)
+{
+	dispatch_release(async_queue_semaphore);
+	free(async_queue);
+}

serve/async.h

@@ -0,0 +1,9 @@
+#ifndef _GUARD_async_h_
+#define _GUARD_async_h_
+
+void dispatch_main_async_f(void* context, void (*cb)(void*));
+void main_async_init(void);
+void main_async_start(EV_P);
+void main_async_destroy(void);
+
+#endif

serve/bbf.c

@@ -0,0 +1,11 @@
+#include "uri.h"
+
+#define MSG ("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 12\r\n\r\nhello world\n")
+
+ebb_buf ebb_bbf_detect_objects(const void* query)
+{
+	ebb_buf buf;
+	buf.data = MSG;
+	buf.len = sizeof(MSG);
+	return buf;
+}

serve/ebb.c

@@ -431,7 +431,7 @@ ebb_connection_init(ebb_connection *connection)
 }
 
 void 
-ebb_connection_schedule_close (ebb_connection *connection)
+ebb_connection_schedule_close(ebb_connection *connection)
 {
   ev_timer_start(connection->server->loop, &connection->goodbye_watcher);
 }
@@ -456,7 +456,7 @@ ebb_connection_reset_timeout(ebb_connection *connection)
  * will return 0 and ignore the request.
  */
 int 
-ebb_connection_write (ebb_connection *connection, const char *buf, size_t len, ebb_after_write_cb cb)
+ebb_connection_write(ebb_connection *connection, const char *buf, size_t len, ebb_after_write_cb cb)
 {
   if(ev_is_active(&connection->write_watcher))
     return 0;

serve/makefile

@@ -1,7 +1,7 @@
 include ../lib/config.mk
 #CC += -faddress-sanitizer -fno-omit-frame-pointer
-LDFLAGS := -L"../lib" -lccv -lev -lBlocksRuntime -ldispatch $(LDFLAGS)
-CFLAGS := -O3 -Wall -fblocks -I"../lib" $(CFLAGS)
+LDFLAGS := -L"../lib" -lccv -lev -ldispatch $(LDFLAGS)
+CFLAGS := -O3 -Wall -I"../lib" $(CFLAGS)
 
 TARGETS = ccv
 
@@ -10,8 +10,8 @@ all: libccv.a $(TARGETS)
 clean:
 	${MAKE} clean -C ../lib ; rm *.o $(TARGETS) -f
 
-$(TARGETS): serve.o ebb.o ebb_request_parser.o libccv.a
-	$(CC) -o $@ serve.o ebb.o ebb_request_parser.o $(LDFLAGS)
+$(TARGETS): serve.o uri.o bbf.o async.o ebb.o ebb_request_parser.o libccv.a
+	$(CC) -o $@ serve.o uri.o bbf.o async.o ebb.o ebb_request_parser.o $(LDFLAGS)
 
 libccv.a:
 	${MAKE} -C ../lib

serve/serve.c

@@ -2,46 +2,79 @@
 #include <ev.h>
 #include <dispatch/dispatch.h>
 #include "ebb.h"
+#include "uri.h"
+#include "async.h"
+
+static const char* ebb_http_404 = "HTTP/1.1 404 Not Found\r\nContent-Type: text/plain\r\nContent-Length: 4\r\n\r\n404\n";
+
+typedef struct {
+	ebb_request* request;
+} ebb_connection_extras;
 
 typedef struct {
 	ebb_connection* connection;
-	ebb_buf* responses;
+	ccv_uri_dispatch_t* dispatcher;
+	char* query;
+	ebb_buf response;
 } ebb_request_extras;
 
-static ev_async main_async;
+static void on_request_path(ebb_request* request, const char* at, size_t length)
+{
+	ebb_request_extras* extras = (ebb_request_extras*)request->data;
+	char* path = (char*)at;
+	char eof = path[length];
+	path[length] = '\0';
+	extras->dispatcher = find_uri_dispatch(path);
+	path[length] = eof;
+}
 
-static void dispatch_main_async(dispatch_block_t block)
+static void on_request_query_string(ebb_request* request, const char* at, size_t length)
 {
-	dispatch_async(dispatch_get_main_queue(), block);
-	ev_async_send(EV_DEFAULT_ &main_async);
+	ebb_request_extras* request_extras = (ebb_request_extras*)request->data;
+	if (request_extras->dispatcher)
+	{
+	}
 }
 
-static void main_async_dispatch(EV_P_ ev_async* w, int revents)
+static void on_connection_response_continue(ebb_connection* connection)
 {
-	dispatch_main();
+	ebb_connection_schedule_close(connection);
 }
 
-static void on_request_path(ebb_request* request, const char* at, size_t length)
+static void on_request_response(void* context)
 {
+	ebb_request* request = (ebb_request*)context;
+	ebb_request_extras* request_extras = (ebb_request_extras*)request->data;
+	ebb_connection* connection = request_extras->connection;
+	ebb_connection_write(connection, request_extras->response.data, request_extras->response.len, on_connection_response_continue);
+	ccfree(request);
 }
 
-static void on_request_query_string(ebb_request* request, const char* at, size_t length)
+static void on_request_processing(void* context)
 {
+	// this is called off-thread
+	ebb_request* request = (ebb_request*)context;
+	ebb_request_extras* request_extras = (ebb_request_extras*)request->data;
+	request_extras->response = request_extras->dispatcher->dispatch(0);
+	dispatch_main_async_f(request, on_request_response);
 }
 
 static void on_request_dispatch(ebb_request* request)
 {
-	dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
-		// do the real computation off thread
-		dispatch_main_async(^{
-		});
-	});
+	ebb_request_extras* request_extras = (ebb_request_extras*)request->data;
+	ebb_connection* connection = request_extras->connection;
+	if (request_extras->dispatcher)
+		dispatch_async_f(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), request, on_request_processing);
+	else // write 404
+		ebb_connection_write(connection, ebb_http_404, sizeof(ebb_http_404), on_connection_response_continue);
 }
 
 static ebb_request* new_request(ebb_connection* connection)
 {
 	ebb_request* request = (ebb_request*)ccmalloc(sizeof(ebb_request) + sizeof(ebb_request_extras));
 	ebb_request_init(request);
+	ebb_connection_extras* connection_extras = (ebb_connection_extras*)(connection->data);
+	connection_extras->request = request;
 	ebb_request_extras* request_extras = (ebb_request_extras*)(request + 1);
 	request_extras->connection = connection;
 	request->data = request_extras;
@@ -58,8 +91,11 @@ static void on_connection_close(ebb_connection* connection)
 
 static ebb_connection* new_connection(ebb_server* server, struct sockaddr_in* addr)
 {
-	ebb_connection* connection = (ebb_connection*)ccmalloc(sizeof(ebb_connection));
+	ebb_connection* connection = (ebb_connection*)ccmalloc(sizeof(ebb_connection) + sizeof(ebb_connection_extras));
 	ebb_connection_init(connection);
+	ebb_connection_extras* connection_extras = (ebb_connection_extras*)(connection + 1);
+	connection_extras->request = 0;
+	connection->data = connection_extras;
 	connection->new_request = new_request;
 	connection->on_close = on_connection_close;
 	return connection;
@@ -71,7 +107,10 @@ int main(int argc, char** argv)
 	ebb_server_init(&server, EV_DEFAULT);
 	server.new_connection = new_connection;
 	ebb_server_listen_on_port(&server, 3350);
-	ev_async_init(&main_async, main_async_dispatch);
+	printf("Listen on 3350, http://localhost:3350/\n");
+	main_async_init();
+	main_async_start(EV_DEFAULT);
 	ev_run(EV_DEFAULT_ 0);
+	main_async_destroy();
 	return 0;
 }

serve/uri.c

@@ -0,0 +1,28 @@
+#include "uri.h"
+#include <string.h>
+
+static const ccv_uri_dispatch_t uri_map[] = {
+	{
+		.uri = "/bbf/detect.objects",
+		.dispatch = ebb_bbf_detect_objects,
+	},
+};
+
+
+ccv_uri_dispatch_t* find_uri_dispatch(const char* path)
+{
+	ccv_uri_dispatch_t* low = (ccv_uri_dispatch_t*)uri_map;
+	ccv_uri_dispatch_t* high = (ccv_uri_dispatch_t*)uri_map + sizeof(uri_map) / sizeof(ccv_uri_dispatch_t) - 1;
+	while (low <= high)
+	{
+		ccv_uri_dispatch_t* middle = low + (high - low) / 2;
+		int flag = strcmp(middle->uri, path);
+		if (flag == 0)
+			return middle;
+		else if (flag < 0)
+			low = middle + 1;
+		else
+			high = middle - 1;
+	}
+	return 0;
+}

serve/uri.h

@@ -0,0 +1,14 @@
+#ifndef _GUARD_uri_h_
+#define _GUARD_uri_h_
+
+#include "ebb.h"
+
+typedef struct {
+	char* uri;
+	ebb_buf (*dispatch)(const void*);
+} ccv_uri_dispatch_t;
+
+ccv_uri_dispatch_t* find_uri_dispatch(const char* path);
+ebb_buf ebb_bbf_detect_objects(const void* query);
+
+#endif

test/functional/makefile

@@ -2,7 +2,7 @@ include ../../lib/config.mk
 
 #CC +=# -fprofile-arcs -ftest-coverage
 LDFLAGS := -L"../../lib" -lccv $(LDFLAGS)
-CFLAGS := -O3 -msse2 -Wall -I"../../lib" -I"../" $(CFLAGS)
+CFLAGS := -O3 -Wall -I"../../lib" -I"../" $(CFLAGS)
 TARGETS = algebra.tests util.tests numeric.tests basic.tests memory.tests io.tests transform.tests
 
 test: all