netrender/slave.py

   1 # ##### BEGIN GPL LICENSE BLOCK #####
   2 #
   3 #  This program is free software; you can redistribute it and/or
   4 #  modify it under the terms of the GNU General Public License
   5 #  as published by the Free Software Foundation; either version 2
   6 #  of the License, or (at your option) any later version.
   7 #
   8 #  This program is distributed in the hope that it will be useful,
   9 #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 #  GNU General Public License for more details.
  12 #
  13 #  You should have received a copy of the GNU General Public License
  14 #  along with this program; if not, write to the Free Software Foundation,
  15 #  Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
  16 #
  17 # ##### END GPL LICENSE BLOCK #####
  18
  19 import sys, os, platform, shutil
  20 import http, http.client, http.server
  21 import subprocess, time, threading
  22 import json
  23
  24 import bpy
  25
  26 from netrender.utils import *
  27 import netrender.model
  28 import netrender.repath
  29 import netrender.baking
  30 import netrender.thumbnail as thumbnail
  31
  32
  33 CANCEL_POLL_SPEED = 2
  34 MAX_TIMEOUT = 10
  35 INCREMENT_TIMEOUT = 1
  36 MAX_CONNECT_TRY = 10
  37
  38 def clearSlave(path):
  39     shutil.rmtree(path)
  40
  41 def slave_Info(netsettings):
  42     sysname, nodename, release, version, machine, processor = platform.uname()
  43     slave = netrender.model.RenderSlave()
  44     slave.name = nodename
  45     slave.stats = sysname + " " + release + " " + machine + " " + processor
  46     if netsettings.slave_tags:
  47         slave.tags = set(netsettings.slave_tags.split(";"))
  48
  49     if netsettings.slave_bake:
  50         slave.tags.add(netrender.model.TAG_BAKING)
  51
  52     if netsettings.slave_render:
  53         slave.tags.add(netrender.model.TAG_RENDER)
  54
  55     return slave
  56
  57 def testCancel(conn, job_id, frame_number):
  58         with ConnectionContext():
  59             conn.request("HEAD", "/status", headers={"job-id":job_id, "job-frame": str(frame_number)})
  60
  61         # canceled if job isn't found anymore
  62         if responseStatus(conn) == http.client.NO_CONTENT:
  63             return True
  64         else:
  65             return False
  66
  67 def testFile(conn, job_id, slave_id, rfile, job_prefix, main_path=None):
  68     job_full_path = createLocalPath(rfile, job_prefix, main_path, rfile.force)
  69
  70     found = os.path.exists(job_full_path)
  71
  72     if found and rfile.signature is not None:
  73         found_signature = hashFile(job_full_path)
  74         found = found_signature == rfile.signature
  75
  76         if not found:
  77             print("Found file %s at %s but signature mismatch!" % (rfile.filepath, job_full_path))
  78             os.remove(job_full_path)
  79
  80     if not found:
  81         # Force prefix path if not found
  82         job_full_path = createLocalPath(rfile, job_prefix, main_path, True)
  83         print("Downloading", job_full_path)
  84         temp_path = os.path.join(job_prefix, "slave.temp")
  85         with ConnectionContext():
  86             conn.request("GET", fileURL(job_id, rfile.index), headers={"slave-id":slave_id})
  87         response = conn.getresponse()
  88
  89         if response.status != http.client.OK:
  90             return None # file for job not returned by server, need to return an error code to server
  91
  92         f = open(temp_path, "wb")
  93         buf = response.read(1024)
  94
  95         while buf:
  96             f.write(buf)
  97             buf = response.read(1024)
  98
  99         f.close()
 100
 101         os.renames(temp_path, job_full_path)
 102
 103     rfile.filepath = job_full_path
 104
 105     return job_full_path
 106
 107 def breakable_timeout(timeout):
 108     for i in range(timeout):
 109         time.sleep(1)
 110         if engine.test_break():
 111             break
 112
 113 def render_slave(engine, netsettings, threads):
 114     bisleep = BreakableIncrementedSleep(INCREMENT_TIMEOUT, 1, MAX_TIMEOUT, engine.test_break)
 115
 116     engine.update_stats("", "Network render node initiation")
 117
 118     slave_path = bpy.path.abspath(netsettings.path)
 119
 120     if not os.path.exists(slave_path):
 121         print("Slave working path ( %s ) doesn't exist" % netsettings.path)
 122         return
 123
 124     if not os.access(slave_path, os.W_OK):
 125         print("Slave working path ( %s ) is not writable" % netsettings.path)
 126         return
 127
 128     conn = clientConnection(netsettings)
 129
 130     if not conn:
 131         print("Connection failed, will try connecting again at most %i times" % MAX_CONNECT_TRY)
 132         bisleep.reset()
 133
 134         for i in range(MAX_CONNECT_TRY):
 135             bisleep.sleep()
 136
 137             conn = clientConnection(netsettings)
 138
 139             if conn or engine.test_break():
 140                 break
 141
 142             print("Retry %i failed, waiting %is before retrying" % (i + 1, bisleep.current))
 143
 144     if conn:
 145         with ConnectionContext():
 146             conn.request("POST", "/slave", json.dumps(slave_Info(netsettings).serialize()))
 147         response = conn.getresponse()
 148         response.read()
 149
 150         slave_id = response.getheader("slave-id")
 151
 152         NODE_PREFIX = os.path.join(slave_path, "slave_" + slave_id)
 153         verifyCreateDir(NODE_PREFIX)
 154
 155         engine.update_stats("", "Network render connected to master, waiting for jobs")
 156
 157         while not engine.test_break():
 158             with ConnectionContext():
 159                 conn.request("GET", "/job", headers={"slave-id":slave_id})
 160             response = conn.getresponse()
 161
 162             if response.status == http.client.OK:
 163                 bisleep.reset()
 164
 165                 job = netrender.model.RenderJob.materialize(json.loads(str(response.read(), encoding='utf8')))
 166                 engine.update_stats("", "Network render processing job from master")
 167
 168                 job_prefix = os.path.join(NODE_PREFIX, "job_" + job.id)
 169                 verifyCreateDir(job_prefix)
 170
 171                 # set tempdir for fsaa temp files
 172                 # have to set environ var because render is done in a subprocess and that's the easiest way to propagate the setting
 173                 os.environ["TMP"] = job_prefix
 174
 175
 176                 if job.type == netrender.model.JOB_BLENDER:
 177                     job_path = job.files[0].original_path # original path of the first file
 178                     main_path, main_file = os.path.split(job_path)
 179
 180                     job_full_path = testFile(conn, job.id, slave_id, job.files[0], job_prefix)
 181                     print("Fullpath", job_full_path)
 182                     print("File:", main_file, "and %i other files" % (len(job.files) - 1,))
 183
 184                     for rfile in job.files[1:]:
 185                         testFile(conn, job.id, slave_id, rfile, job_prefix, main_path)
 186                         print("\t", rfile.filepath)
 187
 188                     netrender.repath.update(job)
 189
 190                     engine.update_stats("", "Render File " + main_file + " for job " + job.id)
 191                 elif job.type == netrender.model.JOB_VCS:
 192                     if not job.version_info:
 193                         # Need to return an error to server, incorrect job type
 194                         pass
 195
 196                     job_path = job.files[0].filepath # path of main file
 197                     main_path, main_file = os.path.split(job_path)
 198
 199                     job.version_info.update()
 200
 201                     # For VCS jobs, file path is relative to the working copy path
 202                     job_full_path = os.path.join(job.version_info.wpath, job_path)
 203
 204                     engine.update_stats("", "Render File " + main_file + " for job " + job.id)
 205
 206                 # announce log to master
 207                 logfile = netrender.model.LogFile(job.id, slave_id, [frame.number for frame in job.frames])
 208                 with ConnectionContext():
 209                     conn.request("POST", "/log", bytes(json.dumps(logfile.serialize()), encoding='utf8'))
 210                 response = conn.getresponse()
 211                 response.read()
 212
 213
 214                 first_frame = job.frames[0].number
 215
 216                 # start render
 217                 start_t = time.time()
 218
 219                 if job.rendersWithBlender():
 220                     frame_args = []
 221
 222                     for frame in job.frames:
 223                         print("frame", frame.number)
 224                         frame_args += ["-f", str(frame.number)]
 225
 226                     with NoErrorDialogContext():
 227                         process = subprocess.Popen(
 228                             [bpy.app.binary_path,
 229                              "-b",
 230                              "-y",
 231                              "-noaudio",
 232                              job_full_path,
 233                              "-t", str(threads),
 234                              "-o", os.path.join(job_prefix, "######"),
 235                              "-E", job.render,
 236                              "-F", "MULTILAYER",
 237                              ] + frame_args,
 238                             stdout=subprocess.PIPE,
 239                             stderr=subprocess.STDOUT,
 240                             )
 241
 242                 elif job.subtype == netrender.model.JOB_SUB_BAKING:
 243                     tasks = []
 244                     for frame in job.frames:
 245                         tasks.append(netrender.baking.commandToTask(frame.command))
 246
 247                     with NoErrorDialogContext():
 248                         process = netrender.baking.bake(job, tasks)
 249
 250                 elif job.type == netrender.model.JOB_PROCESS:
 251                     command = job.frames[0].command
 252                     with NoErrorDialogContext():
 253                         process = subprocess.Popen(command.split(" "), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 254
 255                 headers = {"slave-id":slave_id}
 256
 257                 results = []
 258
 259                 line = ""
 260
 261                 class ProcessData:
 262                     def __init__(self):
 263                         self.lock = threading.Lock()
 264                         self.stdout = bytes()
 265                         self.cancelled = False
 266                         self.start_time = time.time()
 267                         self.last_time = time.time()
 268
 269                 data = ProcessData()
 270
 271                 def run_process(process, data):
 272                     while not data.cancelled and process.poll() is None:
 273                         buf = process.stdout.read(1024)
 274
 275                         data.lock.acquire()
 276                         data.stdout += buf
 277                         data.lock.release()
 278
 279                 process_thread = threading.Thread(target=run_process, args=(process, data))
 280
 281                 process_thread.start()
 282
 283                 while not data.cancelled and process_thread.is_alive():
 284                     time.sleep(CANCEL_POLL_SPEED / 2)
 285                     current_time = time.time()
 286                     data.cancelled = engine.test_break()
 287                     if current_time - data.last_time > CANCEL_POLL_SPEED:
 288
 289                         data.lock.acquire()
 290
 291                         # update logs if needed
 292                         if data.stdout:
 293                             # (only need to update on one frame, they are linked
 294                             with ConnectionContext():
 295                                 conn.request("PUT", logURL(job.id, first_frame), data.stdout, headers=headers)
 296                             responseStatus(conn)
 297
 298                             stdout_text = str(data.stdout, encoding='utf8')
 299
 300                             # Also output on console
 301                             if netsettings.use_slave_output_log:
 302                                 print(stdout_text, end="")
 303
 304                             lines = stdout_text.split("\n")
 305                             lines[0] = line + lines[0]
 306                             line = lines.pop()
 307                             if job.subtype == netrender.model.JOB_SUB_BAKING:
 308                                 results.extend(netrender.baking.resultsFromOuput(lines))
 309
 310                             data.stdout = bytes()
 311
 312                         data.lock.release()
 313
 314                         data.last_time = current_time
 315                         if testCancel(conn, job.id, first_frame):
 316                             engine.update_stats("", "Job canceled by Master")
 317                             data.cancelled = True
 318
 319                 process_thread.join()
 320                 del process_thread
 321
 322                 if job.type == netrender.model.JOB_BLENDER:
 323                     netrender.repath.reset(job)
 324
 325                 if data.cancelled:
 326                     # kill process if needed
 327                     if process.poll() is None:
 328                         try:
 329                             process.terminate()
 330                         except OSError:
 331                             pass
 332                     continue # to next frame
 333
 334                 # read leftovers if needed
 335                 data.stdout += process.stdout.read()
 336
 337                 # flush the rest of the logs
 338                 if data.stdout:
 339                     stdout_text = str(data.stdout, encoding='utf8')
 340
 341                     # Also output on console
 342                     if netsettings.use_slave_output_log:
 343                         print(stdout_text, end="")
 344
 345                     lines = stdout_text.split("\n")
 346                     lines[0] = line + lines[0]
 347                     if job.subtype == netrender.model.JOB_SUB_BAKING:
 348                         results.extend(netrender.baking.resultsFromOuput(lines))
 349
 350                     # (only need to update on one frame, they are linked
 351                     with ConnectionContext():
 352                         conn.request("PUT", logURL(job.id, first_frame), data.stdout, headers=headers)
 353
 354                     if responseStatus(conn) == http.client.NO_CONTENT:
 355                         continue
 356
 357                 total_t = time.time() - data.start_time
 358
 359                 avg_t = total_t / len(job.frames)
 360
 361                 status = process.returncode
 362
 363                 print("status", status)
 364
 365                 headers = {"job-id":job.id, "slave-id":slave_id, "job-time":str(avg_t)}
 366
 367
 368                 if status == 0: # non zero status is error
 369                     headers["job-result"] = str(netrender.model.FRAME_DONE)
 370                     for frame in job.frames:
 371                         headers["job-frame"] = str(frame.number)
 372                         if job.hasRenderResult():
 373                             # send image back to server
 374
 375                             filename = os.path.join(job_prefix, "%06d.exr" % frame.number)
 376
 377                             # thumbnail first
 378                             if netsettings.use_slave_thumb:
 379                                 thumbname = thumbnail.generate(filename)
 380
 381                                 if thumbname:
 382                                     f = open(thumbname, 'rb')
 383                                     with ConnectionContext():
 384                                         conn.request("PUT", "/thumb", f, headers=headers)
 385                                     f.close()
 386                                     responseStatus(conn)
 387
 388                             f = open(filename, 'rb')
 389                             with ConnectionContext():
 390                                 conn.request("PUT", "/render", f, headers=headers)
 391                             f.close()
 392                             if responseStatus(conn) == http.client.NO_CONTENT:
 393                                 continue
 394
 395                         elif job.subtype == netrender.model.JOB_SUB_BAKING:
 396                             index = job.frames.index(frame)
 397
 398                             frame_results = [result_filepath for task_index, result_filepath in results if task_index == index]
 399
 400                             for result_filepath in frame_results:
 401                                 result_path, result_filename = os.path.split(result_filepath)
 402                                 headers["result-filename"] = result_filename
 403                                 headers["job-finished"] = str(result_filepath == frame_results[-1])
 404
 405                                 f = open(result_filepath, 'rb')
 406                                 with ConnectionContext():
 407                                     conn.request("PUT", "/result", f, headers=headers)
 408                                 f.close()
 409                                 if responseStatus(conn) == http.client.NO_CONTENT:
 410                                     continue
 411
 412                         elif job.type == netrender.model.JOB_PROCESS:
 413                             with ConnectionContext():
 414                                 conn.request("PUT", "/render", headers=headers)
 415                             if responseStatus(conn) == http.client.NO_CONTENT:
 416                                 continue
 417                 else:
 418                     headers["job-result"] = str(netrender.model.FRAME_ERROR)
 419                     for frame in job.frames:
 420                         headers["job-frame"] = str(frame.number)
 421                         # send error result back to server
 422                         with ConnectionContext():
 423                             conn.request("PUT", "/render", headers=headers)
 424                         if responseStatus(conn) == http.client.NO_CONTENT:
 425                             continue
 426
 427                 engine.update_stats("", "Network render connected to master, waiting for jobs")
 428             else:
 429                 bisleep.sleep()
 430
 431         conn.close()
 432
 433         if netsettings.use_slave_clear:
 434             clearSlave(NODE_PREFIX)
 435
 436 if __name__ == "__main__":
 437     pass