We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Socket error: address already in use
When running AI:MMO locally, as soon as the main page is loaded, the following error appears in the console. It looks like two workers try to connect to the same port. After this error, the game is executed normally.

`Traceback (most recent call last):
  File "service.py", line 44, in <module>
    run(host=sys.argv[1], port=int(sys.argv[2]), directory=sys.argv[3])
  File "service.py", line 41, in run
    app.run(host, port)
  File "/home/ramon/.local/lib/python2.7/site-packages/flask/app.py", line 841, in run
    run_simple(host, port, self, **options)
  File "/home/ramon/.local/lib/python2.7/site-packages/werkzeug/serving.py", line 739, in run_simple
    inner()
  File "/home/ramon/.local/lib/python2.7/site-packages/werkzeug/serving.py", line 699, in inner
    fd=fd)
  File "/home/ramon/.local/lib/python2.7/site-packages/werkzeug/serving.py", line 593, in make_server
    passthrough_errors, ssl_context, fd=fd)
  File "/home/ramon/.local/lib/python2.7/site-packages/werkzeug/serving.py", line 504, in __init__
    HTTPServer.__init__(self, (host, int(port)), handler)
  File "/usr/lib/python2.7/SocketServer.py", line 417, in __init__
    self.server_bind()
  File "/usr/lib/python2.7/BaseHTTPServer.py", line 108, in server_bind
    SocketServer.TCPServer.server_bind(self)
  File "/usr/lib/python2.7/SocketServer.py", line 431, in server_bind
    self.socket.bind(self.server_address)
  File "/usr/lib/python2.7/socket.py", line 228, in meth
    return getattr(self._sock,name)(*args)
socket.error: [Errno 98] Address already in use
`

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `aimmo-game/simulation/worker_manager.py`
Content:
```
1 import atexit
2 import itertools
3 import json
4 import logging
5 import os
6 import subprocess
7 import tempfile
8 import threading
9 import time
10 
11 import requests
12 from eventlet.greenpool import GreenPool
13 from eventlet.semaphore import Semaphore
14 from pykube import HTTPClient
15 from pykube import KubeConfig
16 from pykube import Pod
17 
18 LOGGER = logging.getLogger(__name__)
19 
20 
21 class _WorkerManagerData(object):
22     """
23     This class is thread safe
24     """
25 
26     def __init__(self, game_state, user_codes):
27         self._game_state = game_state
28         self._user_codes = user_codes
29         self._lock = Semaphore()
30 
31     def _remove_avatar(self, user_id):
32         assert self._lock.locked
33         self._game_state.remove_avatar(user_id)
34         del self._user_codes[user_id]
35 
36     def remove_user_if_code_is_different(self, user):
37         with self._lock:
38             existing_code = self._user_codes.get(user['id'], None)
39             if existing_code != user['code']:
40                 # Remove avatar from the game, so it stops being called for turns
41                 if existing_code is not None:
42                     self._remove_avatar(user['id'])
43                 return True
44             else:
45                 return False
46 
47     def add_avatar(self, user, worker_url):
48         with self._lock:
49             # Add avatar back into game
50             self._game_state.add_avatar(
51                 user_id=user['id'], worker_url="%s/turn/" % worker_url)
52 
53     def set_code(self, user):
54         with self._lock:
55             self._user_codes[user['id']] = user['code']
56 
57     def get_code(self, player_id):
58         with self._lock:
59             return self._user_codes[player_id]
60 
61     def remove_unknown_avatars(self, known_user_ids):
62         with self._lock:
63             unknown_user_ids = set(self._user_codes) - frozenset(known_user_ids)
64             for u in unknown_user_ids:
65                 self._remove_avatar(u)
66             return unknown_user_ids
67 
68     def set_main_avatar(self, avatar_id):
69         with self._lock:
70             self._game_state.main_avatar_id = avatar_id
71 
72 
73 class WorkerManager(threading.Thread):
74     """
75     Methods of this class must be thread safe unless explicitly stated.
76     """
77     daemon = True
78 
79     def __init__(self, game_state, users_url, port=5000):
80         """
81 
82         :param thread_pool:
83         """
84         self._data = _WorkerManagerData(game_state, {})
85         self.users_url = users_url
86         self._pool = GreenPool(size=3)
87         self.port = port
88         super(WorkerManager, self).__init__()
89 
90     def get_code(self, player_id):
91         return self._data.get_code(player_id)
92 
93     def get_persistent_state(self, player_id):
94         """Get the persistent state for a worker."""
95 
96         return None
97 
98     def create_worker(self, player_id):
99         """Create a worker."""
100 
101         raise NotImplemented
102 
103     def remove_worker(self, player_id):
104         """Remove a worker for the given player."""
105 
106         raise NotImplemented
107 
108     # TODO handle failure
109     def spawn(self, user):
110         # Get persistent state from worker
111         persistent_state = self.get_persistent_state(user['id'])  # noqa: F841
112 
113         # Kill worker
114         LOGGER.info("Removing worker for user %s" % user['id'])
115         self.remove_worker(user['id'])
116 
117         self._data.set_code(user)
118 
119         # Spawn worker
120         LOGGER.info("Spawning worker for user %s" % user['id'])
121         worker_url = self.create_worker(user['id'])
122 
123         # Add avatar back into game
124         self._data.add_avatar(user, worker_url)
125         LOGGER.info('Added user %s', user['id'])
126 
127     def _parallel_map(self, func, iterable_args):
128         list(self._pool.imap(func, iterable_args))
129 
130     def update(self):
131         try:
132             LOGGER.info("Waking up")
133             game_data = requests.get(self.users_url).json()
134         except (requests.RequestException, ValueError) as err:
135             LOGGER.error("Failed to obtain game data : %s", err)
136         else:
137             game = game_data['main']
138 
139             # Remove users with different code
140             users_to_add = []
141             for user in game['users']:
142                 if self._data.remove_user_if_code_is_different(user):
143                     users_to_add.append(user)
144             LOGGER.debug("Need to add users: %s" % [x['id'] for x in users_to_add])
145 
146             # Add missing users
147             self._parallel_map(self.spawn, users_to_add)
148 
149             # Delete extra users
150             known_avatars = set(user['id'] for user in game['users'])
151             removed_user_ids = self._data.remove_unknown_avatars(known_avatars)
152             LOGGER.debug("Removing users: %s" % removed_user_ids)
153             self._parallel_map(self.remove_worker, removed_user_ids)
154 
155             # Update main avatar
156             self._data.set_main_avatar(game_data['main']['main_avatar'])
157 
158     def run(self):
159         while True:
160             self.update()
161             LOGGER.info("Sleeping")
162             time.sleep(10)
163 
164 
165 class LocalWorkerManager(WorkerManager):
166     """Relies on them already being created already."""
167 
168     host = '127.0.0.1'
169     worker_directory = os.path.join(
170         os.path.dirname(__file__),
171         '../../aimmo-game-worker/',
172     )
173 
174     def __init__(self, *args, **kwargs):
175         self.workers = {}
176         self.port_counter = itertools.count(1989)
177         super(LocalWorkerManager, self).__init__(*args, **kwargs)
178 
179     def create_worker(self, player_id):
180         assert(player_id not in self.workers)
181         port = self.port_counter.next()
182         env = os.environ.copy()
183         data_dir = tempfile.mkdtemp()
184 
185         LOGGER.debug('Data dir is %s', data_dir)
186         data = requests.get("http://127.0.0.1:{}/player/{}".format(self.port, player_id)).json()
187 
188         options = data['options']
189         with open('{}/options.json'.format(data_dir), 'w') as options_file:
190             json.dump(options, options_file)
191 
192         code = data['code']
193         with open('{}/avatar.py'.format(data_dir), 'w') as avatar_file:
194             avatar_file.write(code)
195 
196         env['PYTHONPATH'] = data_dir
197 
198         process = subprocess.Popen(['python', 'service.py', self.host, str(port), str(data_dir)], cwd=self.worker_directory, env=env)
199         atexit.register(process.kill)
200         self.workers[player_id] = process
201         worker_url = 'http://%s:%d' % (
202             self.host,
203             port,
204         )
205         LOGGER.info("Worker started for %s, listening at %s", player_id, worker_url)
206         return worker_url
207 
208     def remove_worker(self, player_id):
209         if player_id in self.workers:
210             self.workers[player_id].kill()
211             del self.workers[player_id]
212 
213 
214 class KubernetesWorkerManager(WorkerManager):
215     """Kubernetes worker manager."""
216 
217     def __init__(self, *args, **kwargs):
218         self.api = HTTPClient(KubeConfig.from_service_account())
219         self.game_id = os.environ['GAME_ID']
220         self.game_url = os.environ['GAME_URL']
221         super(KubernetesWorkerManager, self).__init__(*args, **kwargs)
222 
223     def create_worker(self, player_id):
224         pod = Pod(
225             self.api,
226             {
227              'kind': 'Pod',
228              'apiVersion': 'v1',
229              'metadata': {
230                 'generateName': "aimmo-%s-worker-%s-" % (self.game_id, player_id),
231                 'labels': {
232                     'app': 'aimmo-game-worker',
233                     'game': self.game_id,
234                     'player': str(player_id),
235                     },
236                 },
237              'spec': {
238                 'containers': [
239                     {
240                         'env': [
241                             {
242                                 'name': 'DATA_URL',
243                                 'value': "%s/player/%d" % (self.game_url, player_id),
244                             },
245                         ],
246                         'name': 'aimmo-game-worker',
247                         'image': 'ocadotechnology/aimmo-game-worker:%s' % os.environ.get('IMAGE_SUFFIX', 'latest'),
248                         'ports': [
249                             {
250                                 'containerPort': 5000,
251                                 'protocol': 'TCP'
252                             }
253                         ],
254                         'resources': {
255                             'limits': {
256                                 'cpu': '10m',
257                                 'memory': '64Mi',
258                             },
259                         },
260                     },
261                 ],
262              },
263             }
264         )
265         pod.create()
266         iterations = 0
267         while pod.obj['status']['phase'] == 'Pending':
268             if iterations > 30:
269                 raise EnvironmentError('Could not start worker %s, details %s' % (player_id, pod.obj))
270             LOGGER.debug('Waiting for worker %s', player_id)
271             time.sleep(5)
272             pod.reload()
273             iterations += 1
274         worker_url = "http://%s:5000" % pod.obj['status']['podIP']
275         LOGGER.info("Worker started for %s, listening at %s", player_id, worker_url)
276         return worker_url
277 
278     def remove_worker(self, player_id):
279         for pod in Pod.objects(self.api).filter(selector={
280             'app': 'aimmo-game-worker',
281             'game': self.game_id,
282             'player': str(player_id),
283         }):
284             LOGGER.debug('Removing pod %s', pod.obj['spec'])
285             pod.delete()
286 
287 WORKER_MANAGERS = {
288     'local': LocalWorkerManager,
289     'kubernetes': KubernetesWorkerManager,
290 }
291
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/aimmo-game/simulation/worker_manager.py b/aimmo-game/simulation/worker_manager.py
--- a/aimmo-game/simulation/worker_manager.py
+++ b/aimmo-game/simulation/worker_manager.py
@@ -173,7 +173,7 @@
 
     def __init__(self, *args, **kwargs):
         self.workers = {}
-        self.port_counter = itertools.count(1989)
+        self.port_counter = itertools.count(self.port + 10)
         super(LocalWorkerManager, self).__init__(*args, **kwargs)
 
     def create_worker(self, player_id):

{"golden_diff": "diff --git a/aimmo-game/simulation/worker_manager.py b/aimmo-game/simulation/worker_manager.py\n--- a/aimmo-game/simulation/worker_manager.py\n+++ b/aimmo-game/simulation/worker_manager.py\n@@ -173,7 +173,7 @@\n \n     def __init__(self, *args, **kwargs):\n         self.workers = {}\n-        self.port_counter = itertools.count(1989)\n+        self.port_counter = itertools.count(self.port + 10)\n         super(LocalWorkerManager, self).__init__(*args, **kwargs)\n \n     def create_worker(self, player_id):\n", "issue": "Socket error: address already in use\nWhen running AI:MMO locally, as soon as the main page is loaded, the following error appears in the console. It looks like two workers try to connect to the same port. After this error, the game is executed normally.\r\n\r\n`Traceback (most recent call last):\r\n  File \"service.py\", line 44, in <module>\r\n    run(host=sys.argv[1], port=int(sys.argv[2]), directory=sys.argv[3])\r\n  File \"service.py\", line 41, in run\r\n    app.run(host, port)\r\n  File \"/home/ramon/.local/lib/python2.7/site-packages/flask/app.py\", line 841, in run\r\n    run_simple(host, port, self, **options)\r\n  File \"/home/ramon/.local/lib/python2.7/site-packages/werkzeug/serving.py\", line 739, in run_simple\r\n    inner()\r\n  File \"/home/ramon/.local/lib/python2.7/site-packages/werkzeug/serving.py\", line 699, in inner\r\n    fd=fd)\r\n  File \"/home/ramon/.local/lib/python2.7/site-packages/werkzeug/serving.py\", line 593, in make_server\r\n    passthrough_errors, ssl_context, fd=fd)\r\n  File \"/home/ramon/.local/lib/python2.7/site-packages/werkzeug/serving.py\", line 504, in __init__\r\n    HTTPServer.__init__(self, (host, int(port)), handler)\r\n  File \"/usr/lib/python2.7/SocketServer.py\", line 417, in __init__\r\n    self.server_bind()\r\n  File \"/usr/lib/python2.7/BaseHTTPServer.py\", line 108, in server_bind\r\n    SocketServer.TCPServer.server_bind(self)\r\n  File \"/usr/lib/python2.7/SocketServer.py\", line 431, in server_bind\r\n    self.socket.bind(self.server_address)\r\n  File \"/usr/lib/python2.7/socket.py\", line 228, in meth\r\n    return getattr(self._sock,name)(*args)\r\nsocket.error: [Errno 98] Address already in use\r\n`\n", "before_files": [{"content": "import atexit\nimport itertools\nimport json\nimport logging\nimport os\nimport subprocess\nimport tempfile\nimport threading\nimport time\n\nimport requests\nfrom eventlet.greenpool import GreenPool\nfrom eventlet.semaphore import Semaphore\nfrom pykube import HTTPClient\nfrom pykube import KubeConfig\nfrom pykube import Pod\n\nLOGGER = logging.getLogger(__name__)\n\n\nclass _WorkerManagerData(object):\n    \"\"\"\n    This class is thread safe\n    \"\"\"\n\n    def __init__(self, game_state, user_codes):\n        self._game_state = game_state\n        self._user_codes = user_codes\n        self._lock = Semaphore()\n\n    def _remove_avatar(self, user_id):\n        assert self._lock.locked\n        self._game_state.remove_avatar(user_id)\n        del self._user_codes[user_id]\n\n    def remove_user_if_code_is_different(self, user):\n        with self._lock:\n            existing_code = self._user_codes.get(user['id'], None)\n            if existing_code != user['code']:\n                # Remove avatar from the game, so it stops being called for turns\n                if existing_code is not None:\n                    self._remove_avatar(user['id'])\n                return True\n            else:\n                return False\n\n    def add_avatar(self, user, worker_url):\n        with self._lock:\n            # Add avatar back into game\n            self._game_state.add_avatar(\n                user_id=user['id'], worker_url=\"%s/turn/\" % worker_url)\n\n    def set_code(self, user):\n        with self._lock:\n            self._user_codes[user['id']] = user['code']\n\n    def get_code(self, player_id):\n        with self._lock:\n            return self._user_codes[player_id]\n\n    def remove_unknown_avatars(self, known_user_ids):\n        with self._lock:\n            unknown_user_ids = set(self._user_codes) - frozenset(known_user_ids)\n            for u in unknown_user_ids:\n                self._remove_avatar(u)\n            return unknown_user_ids\n\n    def set_main_avatar(self, avatar_id):\n        with self._lock:\n            self._game_state.main_avatar_id = avatar_id\n\n\nclass WorkerManager(threading.Thread):\n    \"\"\"\n    Methods of this class must be thread safe unless explicitly stated.\n    \"\"\"\n    daemon = True\n\n    def __init__(self, game_state, users_url, port=5000):\n        \"\"\"\n\n        :param thread_pool:\n        \"\"\"\n        self._data = _WorkerManagerData(game_state, {})\n        self.users_url = users_url\n        self._pool = GreenPool(size=3)\n        self.port = port\n        super(WorkerManager, self).__init__()\n\n    def get_code(self, player_id):\n        return self._data.get_code(player_id)\n\n    def get_persistent_state(self, player_id):\n        \"\"\"Get the persistent state for a worker.\"\"\"\n\n        return None\n\n    def create_worker(self, player_id):\n        \"\"\"Create a worker.\"\"\"\n\n        raise NotImplemented\n\n    def remove_worker(self, player_id):\n        \"\"\"Remove a worker for the given player.\"\"\"\n\n        raise NotImplemented\n\n    # TODO handle failure\n    def spawn(self, user):\n        # Get persistent state from worker\n        persistent_state = self.get_persistent_state(user['id'])  # noqa: F841\n\n        # Kill worker\n        LOGGER.info(\"Removing worker for user %s\" % user['id'])\n        self.remove_worker(user['id'])\n\n        self._data.set_code(user)\n\n        # Spawn worker\n        LOGGER.info(\"Spawning worker for user %s\" % user['id'])\n        worker_url = self.create_worker(user['id'])\n\n        # Add avatar back into game\n        self._data.add_avatar(user, worker_url)\n        LOGGER.info('Added user %s', user['id'])\n\n    def _parallel_map(self, func, iterable_args):\n        list(self._pool.imap(func, iterable_args))\n\n    def update(self):\n        try:\n            LOGGER.info(\"Waking up\")\n            game_data = requests.get(self.users_url).json()\n        except (requests.RequestException, ValueError) as err:\n            LOGGER.error(\"Failed to obtain game data : %s\", err)\n        else:\n            game = game_data['main']\n\n            # Remove users with different code\n            users_to_add = []\n            for user in game['users']:\n                if self._data.remove_user_if_code_is_different(user):\n                    users_to_add.append(user)\n            LOGGER.debug(\"Need to add users: %s\" % [x['id'] for x in users_to_add])\n\n            # Add missing users\n            self._parallel_map(self.spawn, users_to_add)\n\n            # Delete extra users\n            known_avatars = set(user['id'] for user in game['users'])\n            removed_user_ids = self._data.remove_unknown_avatars(known_avatars)\n            LOGGER.debug(\"Removing users: %s\" % removed_user_ids)\n            self._parallel_map(self.remove_worker, removed_user_ids)\n\n            # Update main avatar\n            self._data.set_main_avatar(game_data['main']['main_avatar'])\n\n    def run(self):\n        while True:\n            self.update()\n            LOGGER.info(\"Sleeping\")\n            time.sleep(10)\n\n\nclass LocalWorkerManager(WorkerManager):\n    \"\"\"Relies on them already being created already.\"\"\"\n\n    host = '127.0.0.1'\n    worker_directory = os.path.join(\n        os.path.dirname(__file__),\n        '../../aimmo-game-worker/',\n    )\n\n    def __init__(self, *args, **kwargs):\n        self.workers = {}\n        self.port_counter = itertools.count(1989)\n        super(LocalWorkerManager, self).__init__(*args, **kwargs)\n\n    def create_worker(self, player_id):\n        assert(player_id not in self.workers)\n        port = self.port_counter.next()\n        env = os.environ.copy()\n        data_dir = tempfile.mkdtemp()\n\n        LOGGER.debug('Data dir is %s', data_dir)\n        data = requests.get(\"http://127.0.0.1:{}/player/{}\".format(self.port, player_id)).json()\n\n        options = data['options']\n        with open('{}/options.json'.format(data_dir), 'w') as options_file:\n            json.dump(options, options_file)\n\n        code = data['code']\n        with open('{}/avatar.py'.format(data_dir), 'w') as avatar_file:\n            avatar_file.write(code)\n\n        env['PYTHONPATH'] = data_dir\n\n        process = subprocess.Popen(['python', 'service.py', self.host, str(port), str(data_dir)], cwd=self.worker_directory, env=env)\n        atexit.register(process.kill)\n        self.workers[player_id] = process\n        worker_url = 'http://%s:%d' % (\n            self.host,\n            port,\n        )\n        LOGGER.info(\"Worker started for %s, listening at %s\", player_id, worker_url)\n        return worker_url\n\n    def remove_worker(self, player_id):\n        if player_id in self.workers:\n            self.workers[player_id].kill()\n            del self.workers[player_id]\n\n\nclass KubernetesWorkerManager(WorkerManager):\n    \"\"\"Kubernetes worker manager.\"\"\"\n\n    def __init__(self, *args, **kwargs):\n        self.api = HTTPClient(KubeConfig.from_service_account())\n        self.game_id = os.environ['GAME_ID']\n        self.game_url = os.environ['GAME_URL']\n        super(KubernetesWorkerManager, self).__init__(*args, **kwargs)\n\n    def create_worker(self, player_id):\n        pod = Pod(\n            self.api,\n            {\n             'kind': 'Pod',\n             'apiVersion': 'v1',\n             'metadata': {\n                'generateName': \"aimmo-%s-worker-%s-\" % (self.game_id, player_id),\n                'labels': {\n                    'app': 'aimmo-game-worker',\n                    'game': self.game_id,\n                    'player': str(player_id),\n                    },\n                },\n             'spec': {\n                'containers': [\n                    {\n                        'env': [\n                            {\n                                'name': 'DATA_URL',\n                                'value': \"%s/player/%d\" % (self.game_url, player_id),\n                            },\n                        ],\n                        'name': 'aimmo-game-worker',\n                        'image': 'ocadotechnology/aimmo-game-worker:%s' % os.environ.get('IMAGE_SUFFIX', 'latest'),\n                        'ports': [\n                            {\n                                'containerPort': 5000,\n                                'protocol': 'TCP'\n                            }\n                        ],\n                        'resources': {\n                            'limits': {\n                                'cpu': '10m',\n                                'memory': '64Mi',\n                            },\n                        },\n                    },\n                ],\n             },\n            }\n        )\n        pod.create()\n        iterations = 0\n        while pod.obj['status']['phase'] == 'Pending':\n            if iterations > 30:\n                raise EnvironmentError('Could not start worker %s, details %s' % (player_id, pod.obj))\n            LOGGER.debug('Waiting for worker %s', player_id)\n            time.sleep(5)\n            pod.reload()\n            iterations += 1\n        worker_url = \"http://%s:5000\" % pod.obj['status']['podIP']\n        LOGGER.info(\"Worker started for %s, listening at %s\", player_id, worker_url)\n        return worker_url\n\n    def remove_worker(self, player_id):\n        for pod in Pod.objects(self.api).filter(selector={\n            'app': 'aimmo-game-worker',\n            'game': self.game_id,\n            'player': str(player_id),\n        }):\n            LOGGER.debug('Removing pod %s', pod.obj['spec'])\n            pod.delete()\n\nWORKER_MANAGERS = {\n    'local': LocalWorkerManager,\n    'kubernetes': KubernetesWorkerManager,\n}\n", "path": "aimmo-game/simulation/worker_manager.py"}], "after_files": [{"content": "import atexit\nimport itertools\nimport json\nimport logging\nimport os\nimport subprocess\nimport tempfile\nimport threading\nimport time\n\nimport requests\nfrom eventlet.greenpool import GreenPool\nfrom eventlet.semaphore import Semaphore\nfrom pykube import HTTPClient\nfrom pykube import KubeConfig\nfrom pykube import Pod\n\nLOGGER = logging.getLogger(__name__)\n\n\nclass _WorkerManagerData(object):\n    \"\"\"\n    This class is thread safe\n    \"\"\"\n\n    def __init__(self, game_state, user_codes):\n        self._game_state = game_state\n        self._user_codes = user_codes\n        self._lock = Semaphore()\n\n    def _remove_avatar(self, user_id):\n        assert self._lock.locked\n        self._game_state.remove_avatar(user_id)\n        del self._user_codes[user_id]\n\n    def remove_user_if_code_is_different(self, user):\n        with self._lock:\n            existing_code = self._user_codes.get(user['id'], None)\n            if existing_code != user['code']:\n                # Remove avatar from the game, so it stops being called for turns\n                if existing_code is not None:\n                    self._remove_avatar(user['id'])\n                return True\n            else:\n                return False\n\n    def add_avatar(self, user, worker_url):\n        with self._lock:\n            # Add avatar back into game\n            self._game_state.add_avatar(\n                user_id=user['id'], worker_url=\"%s/turn/\" % worker_url)\n\n    def set_code(self, user):\n        with self._lock:\n            self._user_codes[user['id']] = user['code']\n\n    def get_code(self, player_id):\n        with self._lock:\n            return self._user_codes[player_id]\n\n    def remove_unknown_avatars(self, known_user_ids):\n        with self._lock:\n            unknown_user_ids = set(self._user_codes) - frozenset(known_user_ids)\n            for u in unknown_user_ids:\n                self._remove_avatar(u)\n            return unknown_user_ids\n\n    def set_main_avatar(self, avatar_id):\n        with self._lock:\n            self._game_state.main_avatar_id = avatar_id\n\n\nclass WorkerManager(threading.Thread):\n    \"\"\"\n    Methods of this class must be thread safe unless explicitly stated.\n    \"\"\"\n    daemon = True\n\n    def __init__(self, game_state, users_url, port=5000):\n        \"\"\"\n\n        :param thread_pool:\n        \"\"\"\n        self._data = _WorkerManagerData(game_state, {})\n        self.users_url = users_url\n        self._pool = GreenPool(size=3)\n        self.port = port\n        super(WorkerManager, self).__init__()\n\n    def get_code(self, player_id):\n        return self._data.get_code(player_id)\n\n    def get_persistent_state(self, player_id):\n        \"\"\"Get the persistent state for a worker.\"\"\"\n\n        return None\n\n    def create_worker(self, player_id):\n        \"\"\"Create a worker.\"\"\"\n\n        raise NotImplemented\n\n    def remove_worker(self, player_id):\n        \"\"\"Remove a worker for the given player.\"\"\"\n\n        raise NotImplemented\n\n    # TODO handle failure\n    def spawn(self, user):\n        # Get persistent state from worker\n        persistent_state = self.get_persistent_state(user['id'])  # noqa: F841\n\n        # Kill worker\n        LOGGER.info(\"Removing worker for user %s\" % user['id'])\n        self.remove_worker(user['id'])\n\n        self._data.set_code(user)\n\n        # Spawn worker\n        LOGGER.info(\"Spawning worker for user %s\" % user['id'])\n        worker_url = self.create_worker(user['id'])\n\n        # Add avatar back into game\n        self._data.add_avatar(user, worker_url)\n        LOGGER.info('Added user %s', user['id'])\n\n    def _parallel_map(self, func, iterable_args):\n        list(self._pool.imap(func, iterable_args))\n\n    def update(self):\n        try:\n            LOGGER.info(\"Waking up\")\n            game_data = requests.get(self.users_url).json()\n        except (requests.RequestException, ValueError) as err:\n            LOGGER.error(\"Failed to obtain game data : %s\", err)\n        else:\n            game = game_data['main']\n\n            # Remove users with different code\n            users_to_add = []\n            for user in game['users']:\n                if self._data.remove_user_if_code_is_different(user):\n                    users_to_add.append(user)\n            LOGGER.debug(\"Need to add users: %s\" % [x['id'] for x in users_to_add])\n\n            # Add missing users\n            self._parallel_map(self.spawn, users_to_add)\n\n            # Delete extra users\n            known_avatars = set(user['id'] for user in game['users'])\n            removed_user_ids = self._data.remove_unknown_avatars(known_avatars)\n            LOGGER.debug(\"Removing users: %s\" % removed_user_ids)\n            self._parallel_map(self.remove_worker, removed_user_ids)\n\n            # Update main avatar\n            self._data.set_main_avatar(game_data['main']['main_avatar'])\n\n    def run(self):\n        while True:\n            self.update()\n            LOGGER.info(\"Sleeping\")\n            time.sleep(10)\n\n\nclass LocalWorkerManager(WorkerManager):\n    \"\"\"Relies on them already being created already.\"\"\"\n\n    host = '127.0.0.1'\n    worker_directory = os.path.join(\n        os.path.dirname(__file__),\n        '../../aimmo-game-worker/',\n    )\n\n    def __init__(self, *args, **kwargs):\n        self.workers = {}\n        self.port_counter = itertools.count(self.port + 10)\n        super(LocalWorkerManager, self).__init__(*args, **kwargs)\n\n    def create_worker(self, player_id):\n        assert(player_id not in self.workers)\n        port = self.port_counter.next()\n        env = os.environ.copy()\n        data_dir = tempfile.mkdtemp()\n\n        LOGGER.debug('Data dir is %s', data_dir)\n        data = requests.get(\"http://127.0.0.1:{}/player/{}\".format(self.port, player_id)).json()\n\n        options = data['options']\n        with open('{}/options.json'.format(data_dir), 'w') as options_file:\n            json.dump(options, options_file)\n\n        code = data['code']\n        with open('{}/avatar.py'.format(data_dir), 'w') as avatar_file:\n            avatar_file.write(code)\n\n        env['PYTHONPATH'] = data_dir\n\n        process = subprocess.Popen(['python', 'service.py', self.host, str(port), str(data_dir)], cwd=self.worker_directory, env=env)\n        atexit.register(process.kill)\n        self.workers[player_id] = process\n        worker_url = 'http://%s:%d' % (\n            self.host,\n            port,\n        )\n        LOGGER.info(\"Worker started for %s, listening at %s\", player_id, worker_url)\n        return worker_url\n\n    def remove_worker(self, player_id):\n        if player_id in self.workers:\n            self.workers[player_id].kill()\n            del self.workers[player_id]\n\n\nclass KubernetesWorkerManager(WorkerManager):\n    \"\"\"Kubernetes worker manager.\"\"\"\n\n    def __init__(self, *args, **kwargs):\n        self.api = HTTPClient(KubeConfig.from_service_account())\n        self.game_id = os.environ['GAME_ID']\n        self.game_url = os.environ['GAME_URL']\n        super(KubernetesWorkerManager, self).__init__(*args, **kwargs)\n\n    def create_worker(self, player_id):\n        pod = Pod(\n            self.api,\n            {\n             'kind': 'Pod',\n             'apiVersion': 'v1',\n             'metadata': {\n                'generateName': \"aimmo-%s-worker-%s-\" % (self.game_id, player_id),\n                'labels': {\n                    'app': 'aimmo-game-worker',\n                    'game': self.game_id,\n                    'player': str(player_id),\n                    },\n                },\n             'spec': {\n                'containers': [\n                    {\n                        'env': [\n                            {\n                                'name': 'DATA_URL',\n                                'value': \"%s/player/%d\" % (self.game_url, player_id),\n                            },\n                        ],\n                        'name': 'aimmo-game-worker',\n                        'image': 'ocadotechnology/aimmo-game-worker:%s' % os.environ.get('IMAGE_SUFFIX', 'latest'),\n                        'ports': [\n                            {\n                                'containerPort': 5000,\n                                'protocol': 'TCP'\n                            }\n                        ],\n                        'resources': {\n                            'limits': {\n                                'cpu': '10m',\n                                'memory': '64Mi',\n                            },\n                        },\n                    },\n                ],\n             },\n            }\n        )\n        pod.create()\n        iterations = 0\n        while pod.obj['status']['phase'] == 'Pending':\n            if iterations > 30:\n                raise EnvironmentError('Could not start worker %s, details %s' % (player_id, pod.obj))\n            LOGGER.debug('Waiting for worker %s', player_id)\n            time.sleep(5)\n            pod.reload()\n            iterations += 1\n        worker_url = \"http://%s:5000\" % pod.obj['status']['podIP']\n        LOGGER.info(\"Worker started for %s, listening at %s\", player_id, worker_url)\n        return worker_url\n\n    def remove_worker(self, player_id):\n        for pod in Pod.objects(self.api).filter(selector={\n            'app': 'aimmo-game-worker',\n            'game': self.game_id,\n            'player': str(player_id),\n        }):\n            LOGGER.debug('Removing pod %s', pod.obj['spec'])\n            pod.delete()\n\nWORKER_MANAGERS = {\n    'local': LocalWorkerManager,\n    'kubernetes': KubernetesWorkerManager,\n}\n", "path": "aimmo-game/simulation/worker_manager.py"}]}