def setUp(self):
        self.paths_to_delete = []

        # make size small so tracebacks get truncated
        path, size, offset = 'task_%s' % guid(), 36, 0
        self.paths_to_delete.append(path)

        self.lock = IPCLock(is_slave=0)
        self.task = IbisTaskMessage(self.lock.semaphore_id, path,
                                    offset, size)

        self.mm = SharedMmap(path, size, create=True)
        wire.PackedMessageWriter(self.mm).string('ping')
dataset/ETHPy150Open cloudera/ibis/ibis/tests/test_tasks.py/TestPingPongTask.setUp/original