Understanding How Threading Writes to File in Autopilot

[jmdelahanty]

Hello fellow pilots!

Now that I've finished applying to graduate school, I want to spend my evenings/weekends trying to understand the design of how Autopilot uses threads to write data to HDF5 files. I was taking a multithreading course on Udemy before this application craziness got in the way, but I can't seem to get a handle on how autopilot is capable of piping information into a file while also running all these other tasks.

In short, concurrency is hard for me and I'm having trouble understanding how threading is used to gather "long form" data streams and write it to file.

[cxrodgers]

This is where that magic happens:

autopilot/autopilot/data/subject.py

Line 716 in 04b5968

# spawn thread to accept data

A thread is created that runs the function _data_thread (defined later in the same file). It receives data in a queue, saved as self.data_queue. Data is put into that queue by this line:

autopilot/autopilot/agents/terminal.py

Line 591 in 04b5968

self.subjects[subject_name].save_data(value)

, which I believe runs in the main thread.

Because _data_thread runs in its own thread, if that thread crashes, then you will have an unpleasant scenario where everything seems to be working fine, but nothing is actually written to disk. For that reason, the whole data thread is wrapped in try/except, which is intended to keep it from crashing, although I think a SegmentationFault can still bring it down.

A separate thread is used for plotting, and it receives data in a slightly different way, information is sent directly to that plotting object by its networking handle. That happens in a couple of places. The plotting networking handle begins with "P_" so you can find the relevant lines by searching for that.

[cxrodgers]

Also, consider applying to Emory or Georgia Tech!

[jmdelahanty]

Thank you for pointing me towards how it's accomplished in Autopilot! It would have been great to apply to those programs, but I only applied to 7 programs most of which were in California. If I'm honest with myself, I'm a California kid born and raised and I think I could struggle outside of it maybe...

EDIT: I have something that sorta works! I still get an error at the end about this QT Timer thing, but it looks like it's a warning and not an exception?

from threading import Thread, Event
from queue import Queue
import cv2
from tqdm import tqdm

def capture_video(queue, cap):
    # Loop until event occurs
    while True:
        if event.is_set():
            cap.release()
            break
        # Read a frame from the webcam
        ret, frame = cap.read()
        queue.put(frame)


def display_video(queue, out):
    num_frames = 1000

    for frame in tqdm(range(num_frames), desc="Displaying video", ascii=True):
        # Read a frame from the queue
        frame = queue.get()
        # Display the frame
        cv2.imshow('frame', frame)
        # Write to disk
        out.write(frame)

        if cv2.waitKey(1) & 0xFF == ord('q'):
            print("stop capturing video")
            event.set()
            break
    print("Number of frames reached 1000")
    event.set()

if __name__ == '__main__':

    # Create a Queue object
    queue = Queue()

    # Make an event that can be set for killing the threads
    event = Event()

    cap = cv2.VideoCapture(0)

    # Video writer
    fourcc = cv2.VideoWriter_fourcc(*'XVID')
    out = cv2.VideoWriter('output.avi', fourcc, 20.0, (640, 480))

    # Create a Thread object to capture video
    capture_thread = Thread(target=capture_video, args=(queue, cap))

    # Create a Thread object to display video
    display_thread = Thread(target=display_video, args=(queue, out))

    # Start the threads
    capture_thread.start()
    display_thread.start()

    # Wait for the threads to finish
    capture_thread.join()
    display_thread.join()

I still get:

QObject::startTimer: Timers cannot be started from another thread

At the end but hey at least this is doing something now!

[cxrodgers]

Oh, I see, you're asking general questions about threading, not specifically about Autopilot. Before the edit you said something like, "I can't get capture_video" to ever stop. But why would it stop .. it has an infinite loop in it? I would put the timer inside capture_video and once the timer goes off, use a "return" to end the function and I think that should end the thread.

Otherwise you can kill the thread from outside of that process, but I think that is not best practice, you want to let it end itself, so it can shutdown resources, etc.

If you're generically interested in this kind of thing, check out the multiprocessing module as an alternative. I implemented some other projects with it. Among other benefits, you can control the number of processors that it uses. https://stackoverflow.com/questions/18114285/what-are-the-differences-between-the-threading-and-multiprocessing-modules

Threads can read/write to the same variables/objects in memory, which seems like a benefit, but I think introduces the possibility of certain problems that can be hard to debug. https://en.wikipedia.org/wiki/Race_condition

Good luck with this project and with your applications!

[jmdelahanty]

It started out as questions about Autopilot and then while I was looking more at the code, I realized that if I really wanted to understand the different pieces I needed to actually implement some kind of data streaming using threading, so that's what I'm up to now in the evenings.

It makes sense it wouldn't stop in the initial post since you're right, it was an infinite loop. I hadn't understood what join-ing threads really did until I stumbled across some stack overflow posts and tutorials :p

I started out with multiprocessing for this little experiment but since I saw that Autopilot was using threads for writing multiple data streams to disk I wanted to be adventurous and try it that way. That's a really helpful post explaining the differences! Thanks for linking it!

And so far I haven't run into a race condition but I'm certain that if I was to try and do literally anything else other than this simple "put frame in queue, grab frame from queue" I would blow up my program. Maybe for now I'll switch back to multiprocessing since it'll be an easier way to get my hands on some concurrency for this kind of thing.

Also, from looking at things in the repo, it looks like the video writing is also being handled by threads? I'm still kinda confused about when to use threading/processes for writing data and delivering stimuli. Autopilot seems to use processes for things that can't really be interrupted, like a tone, but then uses threads for other things that also naively seem like they shouldn't be interrupted like delivering a reward through a syringe pump. I can't seem to find a general rule about when each technique is appropriate... I should just re-read the paper.

[jmdelahanty]

Fun update to this thanks to your help here and here!

[jmdelahanty]

I'm messing around this evening with threads again and trying to understand the magic of continuous data recordings. And I just can't seem to understand how it works... I'm pretty confused about how the pi is set to sample pins at the appropriate timing interval. For example, if you just want like 1 sample per second how does Autopilot know how often to sample the GPIO?

[cxrodgers]

I don't think it does that. Mostly it uses "callbacks", which will cause the function you specify to run every time the pin goes high. This uses pigpio triggers, I think. I haven't seen parts of autopilot that sample pins at a specified timing interval (although maybe it exists). You could do something like this by adding a call back to a Timer, and put pigpio code to read the pin in the callback for the Timer. I don't think you will get tight real-time guarantees, so this will only be accurate in the millisecond range, not the sub-millisecond range.