Tag: sysv_ipc

This is a story about the details of a C type leaking into Python.

As part of testing my Python wrapper for SysV IPC, I wrote tests for the time-related attributes of the IPC objects that change when something happens to the object. For instance, when someone sends a message to a queue, the queue’s last_send_time attribute (msg_stime in the C code) is updated to the current time.

I have a hard time imagining many programmers care about these attributes. Knowing the last time someone changed the uid of a message queue, for instance, just doesn’t have many use cases. But they’re part of the SysV IPC API and so I want my package to expose them.

I wrote tests to ensure that each one changed when it was supposed to. The tests failed consistently although the code worked when I tested it “by hand” in an interactive shell. Here’s the relevant portion of a failing test:

 def test_property_last_change_time(self):
     """exercise MessageQueue.last_change_time"""
     original_last_change_time = self.mq.last_change_time
     # This might seem like a no-op, but setting the UID to
     # any value triggers a call to msgctl(...IPC_STAT...)
     # which should set last_change_time.
     self.mq.uid = self.mq.uid
     # Ensure the time actually changed.
     self.assertNotEqual(self.mq.last_change_time,
                         original_last_change_time)

The problem is obvious, right? No, it wasn’t obvious to me, either.

The problem is that in C, a message queue’s last change time (msg_ctime) is of variable type time_t which is typedef-ed as an integral type (int or long) on most (all?) systems. Because the test above executed in less than 1 second, the assertion always failed. Setting self.mq.uid correctly caused an update to the last change time (msg_ctime), it was just being updated to the same value that had been saved in the first line of the test.

My solution was to add a little sleeping, like so –

 def test_property_last_change_time(self):
     """exercise MessageQueue.last_change_time"""
     original_last_change_time = self.mq.last_change_time
     time.sleep(1.1)
     # This might seem like a no-op, but setting the UID to
     # any value triggers a call to msgctl(...IPC_STAT...)
     # which should set last_change_time.
     self.mq.uid = self.mq.uid
     # Ensure the time actually changed.
     self.assertNotEqual(self.mq.last_change_time,
                         original_last_change_time)

That ensured that the value stored in original_last_change_time at the start of the test would differ from self.mq.last_change_time by at least 1 at the end of the test.

Lessons learned: don’t forget about C types, especially when you’re wrapping a C API.