Skip to content

Instantly share code, notes, and snippets.

Created May 30, 2020 07:30
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
What would you like to do?
FSM CR demo alternative to call/cc generator(yield)
有限状态机CR demo
import inspect
import random
import threading
from _thread import _local
import dill as pickle
from functools import partial
import os
import sys
# 全局 状态机无关变量
class LocalSpace(threading.local):
space = None
class Machine:
def __init__(self, id_no, creators, factory_name, use_for):
self.id_no = id_no
self.creators = creators
self.cr_status = random.randint(0, len(self.creators) - 1)
print(f'the cr status: {self.cr_status}')
self.factory_name = factory_name
self.use_for = use_for
self.cache = {}
def generate_work_flow(self):
work_flow = {}
for idx, creator in enumerate(self.creators):
work_flow[idx] = partial(self.say, creator)
return work_flow
def say(self, creator):
idx = self.creators.index(creator)
if self.cr_status == idx: = idx
return -2
message = f'{creator} work in {self.factory_name} to {self.use_for}'
self.cache[creator] = message
self.cache[f'{creator}_machine'] = self
if idx + 1 == len(self.creators):
return -1
return self.creators.index(creator) + 1
def go(self):
magic = Magic(self)
work_flow = self.generate_work_flow()
status = 0
if magic.can_restore:
status = magic.restore()
if self.cr_status == status:
self.cr_status = None
except Exception as e:
print(f'restore failed: {e}')
say_it = work_flow[status]
print(f' = {}')
while True:
new_status = say_it()
if new_status == -1:
if new_status == -2:
status = new_status
say_it = work_flow[status]
print(f' = {}')
class Magic:
def __init__(self, machine):
self.machine = machine = os.path.join('/tmp', self.machine.id_no)
self.mark_frame = None
self.mark_globals = {}
def can_restore(self):
return os.path.exists(
def mark(self):
frame = inspect.currentframe()
self.mark_frame = id(frame)
self.mark_globals = frame.f_globals
def checkpoint(self, status):
with open(, 'wb') as file:
file.write(pickle.dumps({'status': status, 'machine': self.machine, 'diff': self.diff_mark_globals()}))
print('checkpoint success')
def is_local_variable(value):
if type(value) == type and _local in value.__mro__ and value != _local:
return True
return False
def diff_mark_globals(self):
diff = []
frame = inspect.currentframe()
current_globals = frame.f_globals
for name in current_globals:
value = current_globals[name]
if name not in self.mark_globals or self.is_local_variable(value):
diff.append((name, value))
return diff
def restore(self):
with open(, 'rb') as file:
data =
data = pickle.loads(data)
print('restore success')
return data['status']
def update_globals(self, diff):
frame = inspect.currentframe()
for name, value in diff:
frame.f_globals[name] = value
def update_machine(self, machine):
self.machine.cache = machine.cache
if __name__ == '__main__':
Machine('smite0', ['a', 'b', 'c'], 'python', 'idea').go()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment