Changeset 329e26a for benchmark/io/http/main.cfa

Timestamp:

Jun 10, 2022, 5:02:03 PM (2 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, ast-experimental, master, pthread-emulation, qualifiedEnum

Children:

8419b76

Parents:

bf7c7ea

Message:

Re-instated the isolate/multi-cluster option.

File:

• benchmark/io/http/main.cfa (modified) (12 diffs)

benchmark/io/http/main.cfa

@@ -37 +37 @@
 // Globals
 //=============================================================================================
-struct ServerCluster {
-        cluster self;
-        processor    * procs;
-
-};
-
 void ?{}( ServerCluster & this ) {
         (this.self){ "Server Cluster", options.clopts.params };
@@ -56 +50 @@
                 (this.procs[i]){ "Benchmark Processor", this.self };
 
-                int c = 0;
-                int n = 1 + (i % cnt);
-                for(int j = 0; j < CPU_SETSIZE; j++) {
-                        if(CPU_ISSET(j, &fullset)) n--;
-                        if(n == 0) {
-                                c = j;
-                                break;
-                        }
-                }
-                cpu_set_t localset;
-                CPU_ZERO(&localset);
-                CPU_SET(c, &localset);
-                ret = pthread_setaffinity_np(this.procs[i].kernel_thread, sizeof(localset), &localset);
-                if( ret != 0 ) abort | "sched_getaffinity failed with" | ret | strerror( ret );
+                // int c = 0;
+                // int n = 1 + (i % cnt);
+                // for(int j = 0; j < CPU_SETSIZE; j++) {
+                //      if(CPU_ISSET(j, &fullset)) n--;
+                //      if(n == 0) {
+                //              c = j;
+                //              break;
+                //      }
+                // }
+                // cpu_set_t localset;
+                // CPU_ZERO(&localset);
+                // CPU_SET(c, &localset);
+                // ret = pthread_setaffinity_np(this.procs[i].kernel_thread, sizeof(localset), &localset);
+                // if( ret != 0 ) abort | "sched_getaffinity failed with" | ret | strerror( ret );
 
                 #if !defined(__CFA_NO_STATISTICS__)
@@ -85 +79 @@
         #endif
 
-        options.clopts.instance = &this.self;
+        options.clopts.instance[options.clopts.cltr_cnt] = &this.self;
+        options.clopts.cltr_cnt++;
 }
 
@@ -187 +182 @@
                         Acceptor * acceptors = 0p;
                         Q * queues = 0p;
-                        ServerCluster cl;
+                        ServerCluster cl[options.clopts.nclusters];
 
                         if(options.stats) {
                                 stats_thrd = alloc();
-                                (*stats_thrd){ cl.self };
+                                (*stats_thrd){ cl };
                         } else {
                                 stats_thrd = 0p;
@@ -198 +193 @@
                         init_protocol();
                         {
+                                int nacceptors = options.clopts.nprocs * options.clopts.nclusters;
                                 conns = alloc(options.clopts.nworkers);
                                 if(options.socket.reuseport) {
-                                        queues = alloc(options.clopts.nprocs);
-                                        acceptors = anew(options.clopts.nprocs);
-                                        for(i; options.clopts.nprocs) {
+                                        queues = alloc(nacceptors);
+                                        acceptors = alloc(nacceptors);
+                                        sout | "Creating" | nacceptors | "Acceptors";
+                                        for(i; nacceptors) {
+                                                (acceptors[i]){ i % options.clopts.nclusters };
+                                        }
+                                        for(i; nacceptors) {
                                                 (queues[i]){};
                                                 {
@@ -219 +219 @@
                                                         cworkers[i].conn.pipe[0] = fds[pipe_off + (i * 2) + 0];
                                                         cworkers[i].conn.pipe[1] = fds[pipe_off + (i * 2) + 1];
-                                                        cworkers[i].queue = &queues[i % options.clopts.nprocs].q;
+                                                        cworkers[i].queue = &queues[i % nacceptors].q;
                                                         conns[i] = &cworkers[i].conn;
                                                 }
@@ -246 +246 @@
                                         }
                                 }
-
-                                sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors";
+                                sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters";
+                                for(i; options.clopts.nclusters) {
+                                        sout | options.clopts.thrd_cnt[i] | nonl;
+                                }
                                 sout | nl;
                                 {
@@ -274 +276 @@
                                 if(options.socket.reuseport) {
                                         sout | "Notifying connections..." | nonl; flush( sout );
-                                        for(i; options.clopts.nprocs) {
+                                        for(i; nacceptors) {
                                                 acceptors[i].done = true;
                                         }
@@ -283 +285 @@
 
                                         sout | "Shutting down Socket..." | nonl; flush( sout );
-                                        for(i; options.clopts.nprocs) {
+                                        for(i; nacceptors) {
                                                 ret = shutdown( acceptors[i].sockfd, SHUT_RD );
                                                 if( ret < 0 ) {
@@ -292 +294 @@
 
                                         sout | "Closing Socket..." | nonl; flush( sout );
-                                        for(i; options.clopts.nprocs) {
+                                        for(i; nacceptors) {
                                                 ret = close( acceptors[i].sockfd );
                                                 if( ret < 0) {
@@ -301 +303 @@
 
                                         sout | "Stopping accept threads..." | nonl; flush( sout );
-                                        for(i; options.clopts.nprocs) {
+                                        for(i; nacceptors) {
                                                 join(acceptors[i]);
                                         }
@@ -307 +309 @@
 
                                         sout | "Draining worker queues..." | nonl; flush( sout );
-                                        for(i; options.clopts.nprocs) {
+                                        for(i; nacceptors) {
                                                 PendingRead * p = 0p;
                                                 while(p = pop(queues[i].q)) {