mcctrl/mcexec: limit thread pool size when too many threads exist on Linux

executer/include/uprotocol.h

@@ -60,6 +60,7 @@
 #define MCEXEC_UP_SIG_THREAD     0x30a02922
 #define MCEXEC_UP_SYSCALL_THREAD 0x30a02924
 #define MCEXEC_UP_TERMINATE_THREAD 0x30a02925
+#define MCEXEC_UP_GET_NUM_POOL_THREADS  0x30a02926
 
 #define MCEXEC_UP_COPY_FROM_MCK  0x30a03000
 #define MCEXEC_UP_COPY_TO_MCK    0x30a03001

executer/kernel/mcctrl/control.c

@@ -898,6 +898,54 @@ put_and_unlock_out:
 	return ret;
 }
 
+#define THREAD_POOL_PER_CPU_THRESHOLD	(128)
+
+int mcctrl_get_num_pool_threads(ihk_os_t os)
+{
+	struct mcctrl_usrdata *ud = ihk_host_os_get_usrdata(os);
+	struct mcctrl_per_proc_data *ppd = NULL;
+	int hash;
+	unsigned long flags;
+	int nr_threads = 0;
+
+	if (!ud) {
+		return -EINVAL;
+	}
+
+	for (hash = 0; hash < MCCTRL_PER_PROC_DATA_HASH_SIZE; ++hash) {
+
+		read_lock_irqsave(&ud->per_proc_data_hash_lock[hash], flags);
+
+		list_for_each_entry(ppd, &ud->per_proc_data_hash[hash], hash) {
+			struct pid *vpid;
+			struct task_struct *ppd_task;
+
+			vpid = find_vpid(ppd->pid);
+			if (!vpid) {
+				printk("%s: WARNING: couldn't find vpid with PID number %d?\n",
+					__FUNCTION__, ppd->pid);
+				continue;
+			}
+
+			ppd_task = get_pid_task(vpid, PIDTYPE_PID);
+			if (!ppd_task) {
+				printk("%s: WARNING: couldn't find task with PID %d?\n",
+					__FUNCTION__, ppd->pid);
+				continue;
+			}
+
+			nr_threads += get_nr_threads(ppd_task);
+			put_task_struct(ppd_task);
+		}
+
+		read_unlock_irqrestore(&ud->per_proc_data_hash_lock[hash], flags);
+	}
+
+	dprintk("%s: nr_threads: %d, num_online_cpus: %d\n",
+		__FUNCTION__, nr_threads, num_online_cpus());
+	return (nr_threads > (num_online_cpus() * THREAD_POOL_PER_CPU_THRESHOLD));
+}
+
 int mcctrl_add_per_proc_data(struct mcctrl_usrdata *ud, int pid, 
 	struct mcctrl_per_proc_data *ppd)
 {
@@ -2372,6 +2420,9 @@ long __mcctrl_control(ihk_os_t os, unsigned int req, unsigned long arg,
 	case MCEXEC_UP_TERMINATE_THREAD:
 		return mcexec_terminate_thread(os, (unsigned long *)arg, file);
 
+	case MCEXEC_UP_GET_NUM_POOL_THREADS:
+		return mcctrl_get_num_pool_threads(os);
+
 	case MCEXEC_UP_COPY_FROM_MCK:
 		return mcexec_copy_from_mck(os, (unsigned long *)arg);
 

executer/kernel/mcctrl/driver.c

@@ -86,6 +86,7 @@ static struct ihk_os_user_call_handler mcctrl_uchs[] = {
 	{ .request = MCEXEC_UP_SIG_THREAD, .func = mcctrl_ioctl },
 	{ .request = MCEXEC_UP_SYSCALL_THREAD, .func = mcctrl_ioctl },
 	{ .request = MCEXEC_UP_TERMINATE_THREAD, .func = mcctrl_ioctl },
+	{ .request = MCEXEC_UP_GET_NUM_POOL_THREADS, .func = mcctrl_ioctl },
 	{ .request = MCEXEC_UP_DEBUG_LOG, .func = mcctrl_ioctl },
 	{ .request = MCEXEC_UP_COPY_FROM_MCK, .func = mcctrl_ioctl },
 	{ .request = MCEXEC_UP_COPY_TO_MCK, .func = mcctrl_ioctl },

executer/user/mcexec.c

@@ -1256,8 +1256,11 @@ create_worker_thread(pthread_barrier_t *init_ready)
 	struct thread_data_s *tp;
 
 	tp = malloc(sizeof(struct thread_data_s));
-	if (!tp)
+	if (!tp) {
+		fprintf(stderr, "%s: error: allocating thread structure\n",
+			__FUNCTION__);
 		return ENOMEM;
+	}
 	memset(tp, '\0', sizeof(struct thread_data_s));
 	tp->cpu = max_cpuid++;
 	tp->lock = &lock;
@@ -1270,7 +1273,7 @@ create_worker_thread(pthread_barrier_t *init_ready)
 	                      &main_loop_thread_func, tp);
 }
 
-void init_worker_threads(int fd) 
+int init_worker_threads(int fd)
 {
 	int i;
 
@@ -1282,12 +1285,13 @@ void init_worker_threads(int fd)
 		int ret = create_worker_thread(&init_ready);
 
 		if (ret) {
-			printf("ERROR: creating syscall threads(%d)\n", ret);
-			exit(1);
+			printf("ERROR: creating syscall threads (%d), check ulimit?\n", ret);
+			return ret;
 		}
 	}
 
 	pthread_barrier_wait(&init_ready);
+	return 0;
 }
 
 #ifdef ENABLE_MCOVERLAYFS
@@ -2150,7 +2154,11 @@ int main(int argc, char **argv)
 
 	init_sigaction();
 
-	init_worker_threads(fd);
+	if (init_worker_threads(fd) < 0) {
+		perror("worker threads: ");
+		close(fd);
+		return 1;
+	}
 
 	if (ioctl(fd, MCEXEC_UP_START_IMAGE, (unsigned long)desc) != 0) {
 		perror("exec");
@@ -3062,7 +3070,22 @@ gettid_out:
 					goto fork_child_sync_pipe;
 				}
 
-				init_worker_threads(fd);
+				/* Check if we need to limit number of threads in the pool */
+				if ((ret = ioctl(fd, MCEXEC_UP_GET_NUM_POOL_THREADS)) < 0) {
+					fprintf(stderr, "Error: obtaining thread pool count\n");
+				}
+
+				/* Limit number of threads */
+				if (ret == 1) {
+					n_threads = 4;
+				}
+
+				ret = 0;
+				if (init_worker_threads(fd) < 0) {
+					perror("worker threads: ");
+					close(fd);
+					ret = -1;
+				}
 
 fork_child_sync_pipe:
 				sem_post(&fs->sem);