使用Python编写一个模仿CPU工作的程序

今天早上早些时候,在我的Planet Python源中,我读到了一篇有趣的文章"开发CARDIAC:纸板计算机(Developing upwards: CARDIAC: The Cardboard Computer)",它是关于名为Cardiac的纸板计算机的.我的一些追随者和读者应该知道,我有一个名为简单CPU(simple-cpu)的项目,过去的数月我一直工作于此,并且已经发布了源代码.我真的应该给这个项目提供一个合适的许可证,这样,其他人可能更感兴趣,并在他们自己的项目中使用.不管怎样,但愿在这发布之后,我可以完成这件事.

在读完了这篇文章以及它链接的页面后,我受到了一些启发,决定为它编写我自己的模拟器,因为我有编写字节码引擎的经验.我计划着跟随这篇文章继续往前,先写一篇关于汇编器的文章,接下来是关于编译器的文章.这样,通过这些文章,你基本上可以学到,如何用Python为Cardiac创建编译工具集. 在简单CPU(simple-cpu)项目中,我已经编写了一个完整的可工作的汇编器.在内置的游戏中,已经有了可工作的编译器的最初步骤.我也选择Cardiac作为一个验证机器是因为它绝对的简单.不需要复杂的记忆,每个操作码只接受单一的参数,所以它是绝好的学习工具.此外,所有的数据参数都是相同的,不需要检测程序是需要一个寄存器,字符串或者还是内存地址.实际上,只有一个寄存器,累加器.因此,让我们开始吧!我们将基于类来创建,这样包含范围.如果你想尝试的话,你可以简单通过子类来增加新的操作码.首先,我们将集中于初始化例程.这个CPU非常简单,所以我们只需要初始化下面的内容: CPU寄存器, 操作码, 内存空间, 读卡器/输入, 和 打印/tty/输出.
 

class Cardiac(object):
""" This class is the cardiac "CPU". """
def __init__(self):
 self.init_cpu()
 self.reset()
 self.init_mem()
 self.init_reader()
 self.init_output()
def reset(self):
 """  This method resets the CPU's registers to their defaults.  """
 self.pc = 0 #: Program Counter
 self.ir = 0 #: Instruction Register
 self.acc = 0 #: Accumulator
 self.running = False #: Are we running?
def init_cpu(self):
 """  This fancy method will automatically build a list of our opcodes into a hash.  This enables us to build a typical case/select system in Python and also keeps  things more DRY. We could have also used the getattr during the process()  method before, and wrapped it around a try/except block, but that looks  a bit messy. This keeps things clean and simple with a nice one-to-one  call-map.   """
 self.__opcodes = {}
 classes = [self.__class__] #: This holds all the classes and base classes.
 while classes:
  cls = classes.pop() # Pop the classes stack and being
  if cls.__bases__: # Does this class have any base classes?
   classes = classes + list(cls.__bases__)
  for name in dir(cls): # Lets iterate through the names.
   if name[:7] == 'opcode_': # We only want opcodes here.
    try:
     opcode = int(name[7:])
    except ValueError:
     raise NameError('Opcodes must be numeric, invalid opcode: ％s' ％ name[7:])
    self.__opcodes.update({opcode:getattr(self, 'opcode_％s' ％ opcode)})
def init_mem(self):
 """  This method resets the Cardiac's memory space to all blank strings, as per Cardiac specs.  """
 self.mem = ['' for i in range(0,100)]
 self.mem[0] = '001' #: The Cardiac bootstrap operation.
def init_reader(self):
 """  This method initializes the input reader.  """
 self.reader = [] #: This variable can be accessed after initializing the class to provide input data.
def init_output(self):
 """  This method initializes the output deck/paper/printer/teletype/etc...  """
 self.output = []

但愿我写的注释能让你们看明白代码的各部分功能.  也许你已经发现这段代码处理指令集的方法(method)跟 simple-cpu 项目有所不同. 由于它能让开发者根据自己的需求轻松的扩展类库, 我打算在后续的项目中继续使用这种处理方式. 随着我对各部分功能原理的深入理解, 项目也在不断的发展变化. 其实吧,  做这样一个项目真的能让人学到不少东西.  对于精通计算机的人来说 ,  CPU 的工作原理啦, 指令集是怎么处理的啦, 都不是问题啦 .  关键是, 能够按照自己的想法去实现这样一个 CPU 仿真器, 真的很好玩. 根据自己想象中的样子, 亲手打造出这样一台仿真器, 然后看着它屁颠屁颠的运行着, 那叫一个有成就感.

接下来, 我们讲下工具函数(utility functions), 这些函数在很多地方都会用到, 而且允许在子类(subclasses)中重写:
 
   

def read_deck(self, fname):
 """  将指令读到 reader 中.  """
 self.reader = [s.rstrip('\n') for s in open(fname, 'r').readlines()]
 self.reader.reverse()
def fetch(self):
 """  根据指令指针(program pointer) 从内存中读出指令, 然后将指令指针加1.  """
 self.ir = int(self.mem[self.pc])
 self.pc +=1
def get_memint(self, data):
 """  由于我们是以字符串形式(*string* based)保存内存数据的, 要仿真 Cardiac, 就要将字符串转化成整数. 如果是其他存储形式的内存, 如 mmap, 可以根据需要重写本函数.  """
 return int(self.mem[data])
def pad(self, data, length=3):
 """  本函数的功能是像 Cardiac 那样, 在数字的前面补0.  """
 orig = int(data)
 padding = '0'*length
 data = '％s％s' ％ (padding, abs(data))
 if orig < 0:
  return '-'+data[-length:]
 return data[-length:]

本文后面我会另外给大家一段能结合 Mixin classes 使用的代码, 灵活性(pluggable)更强些.  最后就剩下这个处理指令集的方法了:
 
   

def process(self):
 """  本函数只处理一条指令. 默认情况下, 从循环代码(running loop)中调用, 你也可以自己写代码, 以单步调试的方式调用它, 或者使用 time.sleep() 降低执行的速度. 如果想用 TK/GTK/Qt/curses 做的前端界面(frontend), 在另外一个线程中操作, 也可以调用本函数.  """
 self.fetch()
 opcode, data = int(math.floor(self.ir / 100)), self.ir ％ 100
 self.__opcodes[opcode](data)
def opcode_0(self, data):
 """ 输入指令 """
 self.mem[data] = self.reader.pop()
def opcode_1(self, data):
 """ 清除指令 """
 self.acc = self.get_memint(data)
def opcode_2(self, data):
 """ 加法指令 """
 self.acc += self.get_memint(data)
def opcode_3(self, data):
 """ 测试累加器内容指令 """
 if self.acc < 0:
  self.pc = data
def opcode_4(self, data):
 """ 位移指令 """
 x,y = int(math.floor(data / 10)), int(data ％ 10)
 for i in range(0,x):
  self.acc = (self.acc * 10) ％ 10000
 for i in range(0,y):
  self.acc = int(math.floor(self.acc / 10))
def opcode_5(self, data):
 """ 输出指令 """
 self.output.append(self.mem[data])
def opcode_6(self, data):
 """ 存储指令 """
 self.mem[data] = self.pad(self.acc)
def opcode_7(self, data):
 """ 减法指令 """
 self.acc -= self.get_memint(data)
def opcode_8(self, data):
 """ 无条件跳转指令 """
 self.pc = data
def opcode_9(self, data):
 """ 终止, 复位指令 """
 self.reset()
def run(self, pc=None):
 """ 这段代码一直执行到遇到 终止/复位 指令为止. """
 if pc:
  self.pc = pc
 self.running = True
 while self.running:
  self.process()
 print "Output:\n％s" ％ '\n'.join(self.output)
 self.init_output()if __name__ == '__main__':
c = Cardiac()
c.read_deck('deck1.txt')
try:
 c.run()
except:
 print "IR: ％s\nPC: ％s\nOutput: ％s\n" ％ (c.ir, c.pc, '\n'.join(c.output))
 raise

这段是上面提到的, 能在 Mixin 中使用的代码, 我重构过后, 代码如下 :
 

class Memory(object):
""" 本类实现仿真器的虚拟内存空间的各种功能 """
def init_mem(self):
 """  用空白字符串清除 Cardiac 系统内存中的所有数据  """
 self.mem = ['' for i in range(0,100)]
 self.mem[0] = '001' #: 启动 Cardiac 系统.
def get_memint(self, data):
 """  由于我们是以字符串形式(*string* based)保存内存数据的, 要仿真 Cardiac, 就要将字符串转化成整数. 如果是其他存储形式的内存, 如 mmap, 可以根据需要重写本函数.  """
 return int(self.mem[data])
def pad(self, data, length=3):
 """  在数字前面补0  """
 orig = int(data)
 padding = '0'*length
 data = '％s％s' ％ (padding, abs(data))
 if orig < 0:
  return '-'+data[-length:]
 return data[-length:]
class IO(object):
""" 本类实现仿真器的 I/O 功能. To enable alternate methods of input and output, swap this. """
def init_reader(self):
 """  初始化 reader.  """
 self.reader = [] #: 此变量在类初始化后, 可以用来读取输入的数据.
def init_output(self):
 """  初始化诸如： deck/paper/printer/teletype/ 之类的输出功能...  """
 self.output = []
def read_deck(self, fname):
 """  将指令读到 reader 中.  """
 self.reader = [s.rstrip('\n') for s in open(fname, 'r').readlines()]
 self.reader.reverse()
def format_output(self):
 """  格式化虚拟 I/O 设备的输出(output)  """
 return '\n'.join(self.output)
def get_input(self):
 """  获取 IO 的输入(input), 也就是说用 reader 读取数据, 代替原来的 raw_input() .  """
 try:
  return self.reader.pop()
 except IndexError:
  # 如果 reader 遇到文件结束标志(EOF) 就用 raw_input() 代替 reader.
  return raw_input('INP: ')[:3]
def stdout(self, data):
 self.output.append(data)
class CPU(object):
""" 本类模拟 cardiac CPU. """
def __init__(self):
 self.init_cpu()
 self.reset()
 try:
  self.init_mem()
 except AttributeError:
  raise NotImplementedError('You need to Mixin a memory-enabled class.')
 try:
  self.init_reader()
  self.init_output()
 except AttributeError:
  raise NotImplementedError('You need to Mixin a IO-enabled class.')
def reset(self):
 """  用默认值重置 CPU 的寄存器  """
 self.pc = 0 #: 指令指针
 self.ir = 0 #: 指令寄存器
 self.acc = 0 #: 累加器
 self.running = False #: 仿真器的运行状态?
def init_cpu(self):
 """  本函数自动在哈希表中创建指令集. 这样我们就可以使用 case/select 方式调用指令, 同时保持代码简洁. 当然, 在 process(） 中使用 getattr 然后用 try/except 捕捉异常也是可以的, 但是代码看起来就没那么简洁了.  """
 self.__opcodes = {}
 classes = [self.__class__] #: 获取全部类, 包含基类.
 while classes:
  cls = classes.pop() # 把堆栈中的类弹出来
  if cls.__bases__: # 判断有没有基类
   classes = classes + list(cls.__bases__)
  for name in dir(cls): # 遍历名称.
   if name[:7] == 'opcode_': # 只需要把指令读出来即可     try:
     opcode = int(name[7:])
    except ValueError:
     raise NameError('Opcodes must be numeric, invalid opcode: ％s' ％ name[7:])
    self.__opcodes.update({opcode:getattr(self, 'opcode_％s' ％ opcode)})
def fetch(self):
 """  根据指令指针(program pointer) 从内存中读取指令, 然后指令指针加 1.  """
 self.ir = self.get_memint(self.pc)
 self.pc +=1
def process(self):
 """  处理当前指令， 只处理一条. 默认情况下是在循环代码中调用(running loop), 也可以自己写代码, 以单步调试方式调用, 或者利用 time.sleep() 降低执行速度. 在 TK/GTK/Qt/curses 做的界面的线程中调用本函数也是可以的.  """
 self.fetch()
 opcode, data = int(math.floor(self.ir / 100)), self.ir ％ 100
 self.__opcodes[opcode](data)
def opcode_0(self, data):
 """ 输入指令 """
 self.mem[data] = self.get_input()
def opcode_1(self, data):
 """ 清除累加器指令 """
 self.acc = self.get_memint(data)
def opcode_2(self, data):
 """ 加法指令 """
 self.acc += self.get_memint(data)
def opcode_3(self, data):
 """ 测试累加器内容指令 """
 if self.acc < 0:
  self.pc = data
def opcode_4(self, data):
 """ 位移指令 """
 x,y = int(math.floor(data / 10)), int(data ％ 10)
 for i in range(0,x):
  self.acc = (self.acc * 10) ％ 10000
 for i in range(0,y):
  self.acc = int(math.floor(self.acc / 10))
def opcode_5(self, data):
 """ 输出指令 """
 self.stdout(self.mem[data])
def opcode_6(self, data):
 """ 存储指令 """
 self.mem[data] = self.pad(self.acc)
def opcode_7(self, data):
 """ 减法指令 """
 self.acc -= self.get_memint(data)
def opcode_8(self, data):
 """ 无条件跳转指令 """
 self.pc = data
def opcode_9(self, data):
 """ 停止/复位指令"""
 self.reset()
def run(self, pc=None):
 """ 这段代码会一直运行， 直到遇到 halt/reset 指令才停止. """
 if pc:
  self.pc = pc
 self.running = True
 while self.running:
  self.process()
 print "Output:\n％s" ％ self.format_output()
 self.init_output()
class Cardiac(CPU, Memory, IO):
passif __name__ == '__main__':
c = Cardiac()
c.read_deck('deck1.txt')
try:
 c.run()
except:
 print "IR: ％s\nPC: ％s\nOutput: ％s\n" ％ (c.ir, c.pc, c.format_output())
 raise

大家可以从 Developing Upwards: CARDIAC: The Cardboard Computer 中找到本文使用的 deck1.txt .

希望本文能启发大家, 怎么去设计基于类的模块, 插拔性强(pluggable)的 Paython 代码, 以及如何开发 CPU 仿真器.   至于本文 CPU 用到的汇编编译器(assembler) , 会在下一篇文章中教大家.

这段是上面提到的, 能在 Mixin 中使用的代码, 我重构过后, 代码如下 :

class Memory(object):
""" 本类实现仿真器的虚拟内存空间的各种功能 """
def init_mem(self):
 """  用空白字符串清除 Cardiac 系统内存中的所有数据  """
 self.mem = ['' for i in range(0,100)]
 self.mem[0] = '001' #: 启动 Cardiac 系统.
def get_memint(self, data):
 """  由于我们是以字符串形式(*string* based)保存内存数据的, 要仿真 Cardiac, 就要将字符串转化成整数. 如果是其他存储形式的内存, 如 mmap, 可以根据需要重写本函数.  """
 return int(self.mem[data])
def pad(self, data, length=3):
 """  在数字前面补0  """
 orig = int(data)
 padding = '0'*length
 data = '％s％s' ％ (padding, abs(data))
 if orig < 0:
  return '-'+data[-length:]
 return data[-length:]
class IO(object):
""" 本类实现仿真器的 I/O 功能. To enable alternate methods of input and output, swap this. """
def init_reader(self):
 """  初始化 reader.  """
 self.reader = [] #: 此变量在类初始化后, 可以用来读取输入的数据.
def init_output(self):
 """  初始化诸如： deck/paper/printer/teletype/ 之类的输出功能...  """
 self.output = []
def read_deck(self, fname):
 """  将指令读到 reader 中.  """
 self.reader = [s.rstrip('\n') for s in open(fname, 'r').readlines()]
 self.reader.reverse()
def format_output(self):
 """  格式化虚拟 I/O 设备的输出(output)  """
 return '\n'.join(self.output)
def get_input(self):
 """  获取 IO 的输入(input), 也就是说用 reader 读取数据, 代替原来的 raw_input() .  """
 try:
  return self.reader.pop()
 except IndexError:
  # 如果 reader 遇到文件结束标志(EOF) 就用 raw_input() 代替 reader.
  return raw_input('INP: ')[:3]
def stdout(self, data):
 self.output.append(data)
class CPU(object):
""" 本类模拟 cardiac CPU. """
def __init__(self):
 self.init_cpu()
 self.reset()
 try:
  self.init_mem()
 except AttributeError:
  raise NotImplementedError('You need to Mixin a memory-enabled class.')
 try:
  self.init_reader()
  self.init_output()
 except AttributeError:
  raise NotImplementedError('You need to Mixin a IO-enabled class.')
def reset(self):
 """  用默认值重置 CPU 的寄存器  """
 self.pc = 0 #: 指令指针
 self.ir = 0 #: 指令寄存器
 self.acc = 0 #: 累加器
 self.running = False #: 仿真器的运行状态?
def init_cpu(self):
 """  本函数自动在哈希表中创建指令集. 这样我们就可以使用 case/select 方式调用指令, 同时保持代码简洁. 当然, 在 process(） 中使用 getattr 然后用 try/except 捕捉异常也是可以的, 但是代码看起来就没那么简洁了.  """
 self.__opcodes = {}
 classes = [self.__class__] #: 获取全部类, 包含基类.
 while classes:
  cls = classes.pop() # 把堆栈中的类弹出来
  if cls.__bases__: # 判断有没有基类
   classes = classes + list(cls.__bases__)
  for name in dir(cls): # 遍历名称.
   if name[:7] == 'opcode_': # 只需要把指令读出来即可     try:
     opcode = int(name[7:])
    except ValueError:
     raise NameError('Opcodes must be numeric, invalid opcode: ％s' ％ name[7:])
    self.__opcodes.update({opcode:getattr(self, 'opcode_％s' ％ opcode)})
def fetch(self):
 """  根据指令指针(program pointer) 从内存中读取指令, 然后指令指针加 1.  """
 self.ir = self.get_memint(self.pc)
 self.pc +=1
def process(self):
 """  处理当前指令， 只处理一条. 默认情况下是在循环代码中调用(running loop), 也可以自己写代码, 以单步调试方式调用, 或者利用 time.sleep() 降低执行速度. 在 TK/GTK/Qt/curses 做的界面的线程中调用本函数也是可以的.  """
 self.fetch()
 opcode, data = int(math.floor(self.ir / 100)), self.ir ％ 100
 self.__opcodes[opcode](data)
def opcode_0(self, data):
 """ 输入指令 """
 self.mem[data] = self.get_input()
def opcode_1(self, data):
 """ 清除累加器指令 """
 self.acc = self.get_memint(data)
def opcode_2(self, data):
 """ 加法指令 """
 self.acc += self.get_memint(data)
def opcode_3(self, data):
 """ 测试累加器内容指令 """
 if self.acc < 0:
  self.pc = data
def opcode_4(self, data):
 """ 位移指令 """
 x,y = int(math.floor(data / 10)), int(data ％ 10)
 for i in range(0,x):
  self.acc = (self.acc * 10) ％ 10000
 for i in range(0,y):
  self.acc = int(math.floor(self.acc / 10))
def opcode_5(self, data):
 """ 输出指令 """
 self.stdout(self.mem[data])
def opcode_6(self, data):
 """ 存储指令 """
 self.mem[data] = self.pad(self.acc)
def opcode_7(self, data):
 """ 减法指令 """
 self.acc -= self.get_memint(data)
def opcode_8(self, data):
 """ 无条件跳转指令 """
 self.pc = data
def opcode_9(self, data):
 """ 停止/复位指令"""
 self.reset()
def run(self, pc=None):
 """ 这段代码会一直运行， 直到遇到 halt/reset 指令才停止. """
 if pc:
  self.pc = pc
 self.running = True
 while self.running:
  self.process()
 print "Output:\n％s" ％ self.format_output()
 self.init_output()
class Cardiac(CPU, Memory, IO):
passif __name__ == '__main__':
c = Cardiac()
c.read_deck('deck1.txt')
try:
 c.run()
except:
 print "IR: ％s\nPC: ％s\nOutput: ％s\n" ％ (c.ir, c.pc, c.format_output())
 raise

大家可以从Developing Upwards: CARDIAC: The Cardboard Computer 中找到本文使用的 deck1.txt 的代码, 我用的是 从 1 计数到 10 的那个例子 .