#!/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 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 from threading import Lock 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) 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) # Example 2 from concurrent.futures import ThreadPoolExecutor def simulate_pool(pool, grid): next_grid = LockingGrid(grid.height, grid.width) futures = [] for y in range(grid.height): for x in range(grid.width): args = (y, x, grid.get, next_grid.set) future = pool.submit(step_cell, *args) # Fan out futures.append(future) for future in futures: future.result() # 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) 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() with ThreadPoolExecutor(max_workers=10) as pool: for i in range(5): columns.append(str(grid)) grid = simulate_pool(pool, grid) print(columns) # Example 4 try: def game_logic(state, neighbors): raise OSError('Problem with I/O') with ThreadPoolExecutor(max_workers=10) as pool: task = pool.submit(game_logic, ALIVE, 3) task.result() except: logging.exception('Expected') else: assert False