Fwd: Potential deadlock in operf when using --pid

From 2b67533e7cacec028fe4e7429ae4ccbdf7d1f567 Mon Sep 17 00:00:00 2001
From: Rei Odaira <[email protected]>
Date: Fri, 24 Jun 2016 12:06:39 -0500
Subject: [PATCH] Refactor operf_record::_start_recoding_new_thread to include
 read from a pipe and checks

---
 libperf_events/operf_counter.cpp | 50 ++++++++++++++++++++++------------------
 libperf_events/operf_counter.h   |  2 +-
 2 files changed, 28 insertions(+), 24 deletions(-)

diff --git a/libperf_events/operf_counter.cpp b/libperf_events/operf_counter.cpp
index 023f210..e51d108 100644
--- a/libperf_events/operf_counter.cpp
+++ b/libperf_events/operf_counter.cpp
@@ -759,14 +759,26 @@ void operf_record::record_process_info(void)
 	}
 }
 
-int operf_record::_start_recoding_new_thread(pid_t id)
+void operf_record::_check_new_thread(void)
 {
+	pid_t pi;
+	ssize_t len;
 	string err_msg;
 	int rc, fd_for_set_output = -1;
 	struct comm_event ce;
 	u64 sample_id;
 	struct pollfd * old_polldata = poll_data;
 
+	len = read(read_comm_pipe, &pi, sizeof(pi));
+
+	if (len < 0 && errno != EAGAIN) {
+		cverb << vrecord << "Non-fatal error: read_comm_pipe returned too few bytes" << endl;
+	} else if (len != sizeof(pi) || (procs.find(pi) != procs.end())) {
+		return;
+	}
+	// Start profiling this new thread
+	cverb << vrecord << "Start recording for new thread " << pi << endl;
+
 	num_mmaps++;
 	poll_data = new struct pollfd [num_mmaps];
 	// Copy only the existing pollfd objects from the array.  The new pollfd will
@@ -777,7 +789,7 @@ int operf_record::_start_recoding_new_thread(pid_t id)
 	// Make a pseudo comm_event object.  At this point, the
 	// only field we need to set is tid.
 	memset(&ce, 0, sizeof(ce));
-	ce.tid = id;
+	ce.tid = pi;
 	add_process(ce);
 
 	for (unsigned event = 0; event < evts.size(); event++) {
@@ -785,13 +797,13 @@ int operf_record::_start_recoding_new_thread(pid_t id)
 		                                   (!pid_started && !system_wide),
 		                                   callgraph, separate_cpu,
 		                                   false, event));
-		if (op_ctr.perf_event_open(id, -1, this, false) < 0) {
+		if (op_ctr.perf_event_open(pi, -1, this, false) < 0) {
 			sample_id = OP_PERF_NO_SAMPLE_ID;
 			// Send special value to convert process to indicate failure
 			ssize_t len = write(write_comm_pipe, &sample_id, sizeof(sample_id));
 			if (len < 0)
 				perror("Internal error on convert write_comm_pipe");
-			return -1;
+			goto fail;
 		}
 		perfCounters.push_back(op_ctr);
 		int fd = op_ctr.get_fd();
@@ -801,15 +813,15 @@ int operf_record::_start_recoding_new_thread(pid_t id)
 		} else {
 			if ((rc = ioctl(fd, PERF_EVENT_IOC_SET_OUTPUT,
 			                fd_for_set_output)) < 0)
-				perror("_start_recoding_new_thread: ioctl #1 failed");
+				perror("_check_new_thread: ioctl #1 failed");
 		}
 
 		if (rc < 0)
-			return rc;
+			goto fail;
 
 		if ((rc = ioctl(fd, PERF_EVENT_IOC_ENABLE)) < 0) {
-			perror("_start_recoding_new_thread: ioctl #2 failed");
-			return rc;
+			perror("_check_new_thread: ioctl #2 failed");
+			goto fail;
 		}
 
 		sample_id = opHeader.h_attrs[event].ids.back();
@@ -822,7 +834,12 @@ int operf_record::_start_recoding_new_thread(pid_t id)
 			cverb << vrecord << "Sent sample_id " << sample_id << " to convert process" << endl;
 	}
 
-	return 0;
+	return;
+
+ fail:
+	cerr << "Unable to collect samples for forked process " << pi
+	     << ". Process may have ended before recording could be started." << endl;
+	return;
 }
 
 
@@ -838,8 +855,6 @@ void operf_record::recordPerfData(void)
 	cerr << "operf: Profiler started" << endl;
 	while (1) {
 		int prev = sample_reads;
-		pid_t pi;
-		ssize_t len;
 
 		for (size_t i = 0; i < samples_array.size(); i++) {
 			if (samples_array[i].base)
@@ -852,18 +867,7 @@ void operf_record::recordPerfData(void)
 			(void)poll(poll_data, poll_count, -1);
 		}
 		if (!quit && track_new_forks && procs.size() > 1) {
-			len = read(read_comm_pipe, &pi, sizeof(pi));
-
-			if (len < 0 && errno != EAGAIN) {
-				cverb << vrecord << "Non-fatal error: read_comm_pipe returned too few bytes" << endl;
-			} else if (len == sizeof(pi) && (procs.find(pi) == procs.end())) {
-				// Start profiling this new thread
-				cverb << vrecord << "Start recording for new thread " << pi << endl;
-				// Don't treat as fatal error if it doesn't work
-				if (_start_recoding_new_thread(pi) < 0)
-					cerr << "Unable to collect samples for forked process " << pi
-					     << ". Process may have ended before recording could be started." << endl;
-			}
+			_check_new_thread();
 		}
 
 		if (quit) {
diff --git a/libperf_events/operf_counter.h b/libperf_events/operf_counter.h
index 6e19da9..99d23ae 100644
--- a/libperf_events/operf_counter.h
+++ b/libperf_events/operf_counter.h
@@ -99,7 +99,7 @@ private:
 	void setup(void);
 	int prepareToRecord(void);
 	int _prepare_to_record_one_fd(int idx, int fd);
-	int _start_recoding_new_thread(pid_t id);
+	void _check_new_thread(void);
 	void record_process_info(void);
 	void write_op_header_info(void);
 	int _write_header_to_file(void);
-- 
2.5.4 (Apple Git-61)

From f47e184e328549b31c32c80a1b749204cc30b0b7 Mon Sep 17 00:00:00 2001
From: Rei Odaira <[email protected]>
Date: Fri, 24 Jun 2016 18:13:11 -0500
Subject: [PATCH] Avoid deadlock between operf-record and operf-read by making
 the sample output pipe of operf-read non-blocking and polling both the output
 pipe and the read comm pipe when the pipe is full

---
 libperf_events/operf_counter.cpp | 50 ++++++++++++++++++++++++++++++++++------
 libperf_events/operf_counter.h   |  1 +
 libperf_events/operf_utils.cpp   | 31 +++++++++++++++++++------
 libperf_events/operf_utils.h     |  2 +-
 4 files changed, 69 insertions(+), 15 deletions(-)

diff --git a/libperf_events/operf_counter.cpp b/libperf_events/operf_counter.cpp
index e51d108..ca2e421 100644
--- a/libperf_events/operf_counter.cpp
+++ b/libperf_events/operf_counter.cpp
@@ -361,6 +361,18 @@ operf_record::operf_record(int out_fd, bool sys_wide, pid_t the_pid, bool pid_ru
 		      << strerror(errno) << endl;
 		_exit(EXIT_FAILURE);
 	}
+	if (track_new_forks && !write_to_file) {
+		/* The pipe of out_fd can become full when the operf-read process
+		   cannot consume it because the operf-read process waits on another pipe.
+		   To avoid this deadlock, make out_fd non-blocking.
+		   See the comments in OP_perf_utils::op_write_output().
+		 */
+		if (fcntl(out_fd, F_SETFL, O_NONBLOCK) < 0) {
+			cverb << vrecord << "operf_record ctor: failed to make out_fd non_blocking; errno is: "
+			      << strerror(errno) << endl;
+			cverb << vrecord << "  continue execution but the operf-record and operf-read processes can deadlock on each other" << endl;
+		}
+	}
 	cverb << vrecord << "calling setup" << endl;
 	setup();
 }
@@ -380,7 +392,7 @@ int operf_record::_write_header_to_file(void)
 		if (opHeader.h_attrs[i].id_offset == (off_t)-1)
 			goto err_out;
 		total += op_write_output(output_fd, &opHeader.h_attrs[i].ids[0],
-		                         opHeader.h_attrs[i].ids.size() * sizeof(u64));
+		                         opHeader.h_attrs[i].ids.size() * sizeof(u64), this);
 	}
 
 	opHeader.attr_offset = lseek(output_fd, 0, SEEK_CUR);
@@ -392,7 +404,7 @@ int operf_record::_write_header_to_file(void)
 		f_attr.attr = attr.attr;
 		f_attr.ids.offset = attr.id_offset;
 		f_attr.ids.size = attr.ids.size() * sizeof(u64);
-		total += op_write_output(output_fd, &f_attr, sizeof(f_attr));
+		total += op_write_output(output_fd, &f_attr, sizeof(f_attr), this);
 	}
 
 	opHeader.data_offset = lseek(output_fd, 0, SEEK_CUR);
@@ -410,7 +422,7 @@ int operf_record::_write_header_to_file(void)
 
 	if (lseek(output_fd, 0, SEEK_SET) == (off_t)-1)
 		goto err_out;
-	total += op_write_output(output_fd, &f_header, sizeof(f_header));
+	total += op_write_output(output_fd, &f_header, sizeof(f_header), this);
 	if (lseek(output_fd, opHeader.data_offset + opHeader.data_size, SEEK_SET) == (off_t)-1)
 		goto err_out;
 	return total;
@@ -433,18 +445,18 @@ int operf_record::_write_header_to_pipe(void)
 	f_header.attrs.size = evts.size() * sizeof(f_attr);
 	f_header.data.size = 0;
 
-	total = op_write_output(output_fd, &f_header, sizeof(f_header));
+	total = op_write_output(output_fd, &f_header, sizeof(f_header), this);
 
 	for (unsigned i = 0; i < evts.size(); i++) {
 		struct op_header_evt_info attr = opHeader.h_attrs[i];
 		f_attr.attr = attr.attr;
 		f_attr.ids.size = attr.ids.size() * sizeof(u64);
-		total += op_write_output(output_fd, &f_attr, sizeof(f_attr));
+		total += op_write_output(output_fd, &f_attr, sizeof(f_attr), this);
 	}
 
 	for (unsigned i = 0; i < evts.size(); i++) {
 		total += op_write_output(output_fd, &opHeader.h_attrs[i].ids[0],
-		                         opHeader.h_attrs[i].ids.size() * sizeof(u64));
+		                         opHeader.h_attrs[i].ids.size() * sizeof(u64), this);
 	}
 	return total;
 }
@@ -744,7 +756,7 @@ void operf_record::record_process_info(void)
 	for (unsigned int proc_idx = 0; proc_idx < procs.size(); proc_idx++, proc_it++)
 	{
 		struct comm_event ce = proc_it->second;
-		int num = OP_perf_utils::op_write_output(output_fd, &ce, ce.header.size);
+		int num = OP_perf_utils::op_write_output(output_fd, &ce, ce.header.size, this);
 		add_to_total(num);
 		if (cverb << vrecord)
 			cout << "Created COMM event for " << ce.comm << endl;
@@ -842,6 +854,30 @@ void operf_record::_check_new_thread(void)
 	return;
 }
 
+void operf_record::handle_output_full(int output)
+{
+	struct pollfd poll_output_and_comm[2];
+	poll_output_and_comm[0].fd = output;
+	poll_output_and_comm[0].events = POLLOUT;
+	poll_output_and_comm[1].fd = read_comm_pipe;
+	poll_output_and_comm[1].events = POLLIN;
+
+	for ( ; ; ) {
+		if (poll(poll_output_and_comm, 2, -1) < 0) {
+			if (errno == EINTR)
+				continue;
+			string errmsg = "Internal error:  Failed to poll output fd and communication pipe. errno is ";
+			errmsg += strerror(errno);
+			throw runtime_error(errmsg);
+		}
+		if ((poll_output_and_comm[1].revents | POLLIN) && procs.size() > 1) {
+			_check_new_thread();
+		}
+		if (poll_output_and_comm[0].revents | POLLOUT)
+			break;
+	}
+}
+
 
 void operf_record::recordPerfData(void)
 {
diff --git a/libperf_events/operf_counter.h b/libperf_events/operf_counter.h
index 99d23ae..e9aa03d 100644
--- a/libperf_events/operf_counter.h
+++ b/libperf_events/operf_counter.h
@@ -93,6 +93,7 @@ public:
 	unsigned int get_total_bytes_recorded(void) const { return total_bytes_recorded; }
 	void register_perf_event_id(unsigned counter, u64 id, perf_event_attr evt_attr);
 	bool get_valid(void) { return valid; }
+	void handle_output_full(int output);
 
 private:
 	void create(std::string outfile, std::vector<operf_event_t> & evts);
diff --git a/libperf_events/operf_utils.cpp b/libperf_events/operf_utils.cpp
index ff972d4..39b5b00 100644
--- a/libperf_events/operf_utils.cpp
+++ b/libperf_events/operf_utils.cpp
@@ -982,7 +982,7 @@ int OP_perf_utils::op_mmap_trace_file(struct mmap_info & info, bool init)
 }
 
 
-int OP_perf_utils::op_write_output(int output, void *buf, size_t size)
+int OP_perf_utils::op_write_output(int output, void *buf, size_t size, operf_record * pr)
 {
 	int sum = 0;
 	while (size) {
@@ -991,6 +991,23 @@ int OP_perf_utils::op_write_output(int output, void *buf, size_t size)
 		if (ret < 0) {
 			if (errno == EINTR)
 				continue;
+			else if (errno == EAGAIN) {
+				/* The output pipe is full.  It is typically because
+				   the consumer of the pipe, the operf-read process, is
+				   being blocked on a read from the comm pipe.
+				   handle_output_full() will poll both the sample output pipe
+				   and the read end of the comm pipe.
+				   If there is a message in the comm pipe from the operf-read
+				   process, handle_output_full() will call _check_new_thread()
+				   to handle the message and then will write a response
+				   to the comm pipe.  As a result, the operf-read process will
+				   finish being blocked on the read from the comm pipe and
+				   consume the output file, and eventually this operf-record
+				   process can write to the output pipe.
+				 */
+				pr->handle_output_full(output);
+				continue;
+			}
 
 			string errmsg = "Internal error:  Failed to write sample data to output fd. errno is ";
 			errmsg += strerror(errno);
@@ -1059,7 +1076,7 @@ void OP_perf_utils::op_get_vsyscall_mapping(pid_t tgid, int output_fd, operf_rec
 			mmap.tid = tgid;
 			mmap.header.size = (sizeof(mmap) -
 					(sizeof(mmap.filename) - size));
-			int num = OP_perf_utils::op_write_output(output_fd, &mmap, mmap.header.size);
+			int num = OP_perf_utils::op_write_output(output_fd, &mmap, mmap.header.size, pr);
 			if (cverb << vrecord)
 				cout << "Created MMAP event for " << imagename << endl;
 			pr->add_to_total(num);
@@ -1130,7 +1147,7 @@ void OP_perf_utils::op_record_process_exec_mmaps(pid_t pid, pid_t tgid, int outp
 			mmap.tid = pid;
 			mmap.header.size = (sizeof(mmap) -
 					(sizeof(mmap.filename) - size));
-			int num = OP_perf_utils::op_write_output(output_fd, &mmap, mmap.header.size);
+			int num = OP_perf_utils::op_write_output(output_fd, &mmap, mmap.header.size, pr);
 			if (cverb << vrecord)
 				cout << "Created MMAP event for " << imagename << endl;
 			pr->add_to_total(num);
@@ -1341,7 +1358,7 @@ static void _record_module_info(int output_fd, operf_record * pr)
 		mmap.tid = 0;
 		mmap.header.size = (sizeof(mmap) -
 				(sizeof(mmap.filename) - size));
-		int num = OP_perf_utils::op_write_output(output_fd, &mmap, mmap.header.size);
+		int num = OP_perf_utils::op_write_output(output_fd, &mmap, mmap.header.size, pr);
 		if (cverb << vrecord)
 			cout << "Created MMAP event for " << module_name << ". Size: "
 			      << module_size << "; start addr: " << start_address << endl;
@@ -1388,7 +1405,7 @@ void OP_perf_utils::op_record_kernel_info(string vmlinux_file, u64 start_addr, u
 	mmap.tid = 0;
 	mmap.header.size = (sizeof(mmap) -
 			(sizeof(mmap.filename) - size));
-	int num = op_write_output(output_fd, &mmap, mmap.header.size);
+	int num = op_write_output(output_fd, &mmap, mmap.header.size, pr);
 	if (cverb << vrecord) {
 		ostringstream message;
 		message << "Created MMAP event of size " << mmap.header.size << " for " <<mmap.filename << ". length: "
@@ -1434,13 +1451,13 @@ void OP_perf_utils::op_get_kernel_event_data(struct mmap_data *md, operf_record
 		buf = &data[old & md->mask];
 		size = md->mask + 1 - (old & md->mask);
 		old += size;
-		pr->add_to_total(op_write_output(out_fd, buf, size));
+		pr->add_to_total(op_write_output(out_fd, buf, size, pr));
 	}
 
 	buf = &data[old & md->mask];
 	size = head - old;
 	old += size;
-	pr->add_to_total(op_write_output(out_fd, buf, size));
+	pr->add_to_total(op_write_output(out_fd, buf, size, pr));
 	md->prev = old;
 	pc->data_tail = old;
 }
diff --git a/libperf_events/operf_utils.h b/libperf_events/operf_utils.h
index 32954cc..e0a3326 100644
--- a/libperf_events/operf_utils.h
+++ b/libperf_events/operf_utils.h
@@ -84,7 +84,7 @@ void op_perfrecord_sigusr1_handler(int sig __attribute__((unused)),
 int op_get_process_info(bool system_wide, pid_t pid, operf_record * pr);
 void op_record_process_exec_mmaps(pid_t pid, pid_t tgid, int output_fd, operf_record * pr);
 void op_get_vsyscall_mapping(pid_t tgid, int output_fd, operf_record * pr);
-int op_write_output(int output, void *buf, size_t size);
+int op_write_output(int output, void *buf, size_t size, operf_record * pr);
 int op_write_event(event_t * event, u64 sample_type);
 int op_read_from_stream(std::ifstream & is, char * buf, std::streamsize sz);
 int op_mmap_trace_file(struct mmap_info & info, bool init);
-- 
2.5.4 (Apple Git-61)