#!/usr/bin/env PYTHONHASHSEED=1234 python3 # Copyright 2014-2019 Brett Slatkin, Pearson Education Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # Reproduce book environment import random random.seed(1234) import logging from pprint import pprint from sys import stdout as STDOUT # Write all output to a temporary directory import atexit import gc import io import os import tempfile TEST_DIR = tempfile.TemporaryDirectory() atexit.register(TEST_DIR.cleanup) # Make sure Windows processes exit cleanly OLD_CWD = os.getcwd() atexit.register(lambda: os.chdir(OLD_CWD)) os.chdir(TEST_DIR.name) def close_open_files(): everything = gc.get_objects() for obj in everything: if isinstance(obj, io.IOBase): obj.close() atexit.register(close_open_files) # Example 1 from threading import Lock ALIVE = '*' EMPTY = '-' class Grid: def __init__(self, height, width): self.height = height self.width = width self.rows = [] for _ in range(self.height): self.rows.append([EMPTY] * self.width) def get(self, y, x): return self.rows[y % self.height][x % self.width] def set(self, y, x, state): self.rows[y % self.height][x % self.width] = state def __str__(self): output = '' for row in self.rows: for cell in row: output += cell output += '\n' return output class LockingGrid(Grid): def __init__(self, height, width): super().__init__(height, width) self.lock = Lock() def __str__(self): with self.lock: return super().__str__() def get(self, y, x): with self.lock: return super().get(y, x) def set(self, y, x, state): with self.lock: return super().set(y, x, state) # Example 2 from threading import Thread def count_neighbors(y, x, get): n_ = get(y - 1, x + 0) # North ne = get(y - 1, x + 1) # Northeast e_ = get(y + 0, x + 1) # East se = get(y + 1, x + 1) # Southeast s_ = get(y + 1, x + 0) # South sw = get(y + 1, x - 1) # Southwest w_ = get(y + 0, x - 1) # West nw = get(y - 1, x - 1) # Northwest neighbor_states = [n_, ne, e_, se, s_, sw, w_, nw] count = 0 for state in neighbor_states: if state == ALIVE: count += 1 return count def game_logic(state, neighbors): # Do some blocking input/output in here: data = my_socket.recv(100) def game_logic(state, neighbors): if state == ALIVE: if neighbors < 2: return EMPTY # Die: Too few elif neighbors > 3: return EMPTY # Die: Too many else: if neighbors == 3: return ALIVE # Regenerate return state def step_cell(y, x, get, set): state = get(y, x) neighbors = count_neighbors(y, x, get) next_state = game_logic(state, neighbors) set(y, x, next_state) def simulate_threaded(grid): next_grid = LockingGrid(grid.height, grid.width) threads = [] for y in range(grid.height): for x in range(grid.width): args = (y, x, grid.get, next_grid.set) thread = Thread(target=step_cell, args=args) thread.start() # Fan out threads.append(thread) for thread in threads: thread.join() # Fan in return next_grid # Example 3 class ColumnPrinter: def __init__(self): self.columns = [] def append(self, data): self.columns.append(data) def __str__(self): row_count = 1 for data in self.columns: row_count = max( row_count, len(data.splitlines()) + 1) rows = [''] * row_count for j in range(row_count): for i, data in enumerate(self.columns): line = data.splitlines()[max(0, j - 1)] if j == 0: padding = ' ' * (len(line) // 2) rows[j] += padding + str(i) + padding else: rows[j] += line if (i + 1) < len(self.columns): rows[j] += ' | ' return '\n'.join(rows) grid = LockingGrid(5, 9) # Changed grid.set(0, 3, ALIVE) grid.set(1, 4, ALIVE) grid.set(2, 2, ALIVE) grid.set(2, 3, ALIVE) grid.set(2, 4, ALIVE) columns = ColumnPrinter() for i in range(5): columns.append(str(grid)) grid = simulate_threaded(grid) # Changed print(columns) # Example 4 def game_logic(state, neighbors): raise OSError('Problem with I/O') # Example 5 import contextlib import io fake_stderr = io.StringIO() with contextlib.redirect_stderr(fake_stderr): thread = Thread(target=game_logic, args=(ALIVE, 3)) thread.start() thread.join() print(fake_stderr.getvalue())