Removing old Tornado asynchronous annotations from Heron shell (#3781)

heron/shell/src/python/handlers/browsehandler.py

@@ -33,16 +33,14 @@ class BrowseHandler(tornado.web.RequestHandler):
   """
 
   # pylint: disable=attribute-defined-outside-init
-  @tornado.web.asynchronous
-  def get(self, path):
+  async def get(self, path):
     ''' get method '''
     if not path:
       path = "."
 
     if not utils.check_path(path):
-      self.write("Only relative paths are allowed")
       self.set_status(403)
-      self.finish()
+      await self.finish("Only relative paths are allowed")
       return
 
     t = Template(utils.get_asset("browse.html"))
@@ -55,5 +53,4 @@ def get(self, path):
         os=os,
         css=utils.get_asset("bootstrap.css")
     )
-    self.write(t.generate(**args))
-    self.finish()
+    await self.finish(t.generate(**args))

heron/shell/src/python/handlers/downloadhandler.py

@@ -20,19 +20,19 @@
 
 
 ''' downloadhandler.py '''
+import mimetypes
 import os
 import logging
-import tornado.web
+from tornado import web, iostream, gen
 import anticrlf
 
 from heron.shell.src.python import utils
 
-class DownloadHandler(tornado.web.RequestHandler):
+class DownloadHandler(web.RequestHandler):
   """
   Responsible for downloading the files.
   """
-  @tornado.web.asynchronous
-  def get(self, path):
+  async def get(self, path):
     """ get method """
 
     handler = logging.StreamHandler()
@@ -43,41 +43,37 @@ def get(self, path):
 
     logger.debug("request to download: %s", path)
 
-    # If the file is large, we want to abandon downloading
-    # if user cancels the requests.
-    # pylint: disable=attribute-defined-outside-init
-    self.connection_closed = False
-
     self.set_header("Content-Disposition", "attachment")
     if not utils.check_path(path):
-      self.write("Only relative paths are allowed")
       self.set_status(403)
-      self.finish()
+      await self.finish("Only relative paths are allowed")
       return
 
     if path is None or not os.path.isfile(path):
-      self.write("File %s  not found" % path)
       self.set_status(404)
-      self.finish()
+      await self.finish("File %s  not found" % path)
       return
 
-    length = int(4 * 1024 * 1024)
-    offset = int(0)
-    while True:
-      data = utils.read_chunk(path, offset=offset, length=length, escape_data=False)
-      if self.connection_closed or 'data' not in data or len(data['data']) < length:
-        break
-      offset += length
-      self.write(data['data'])
-      self.flush()
-
-    if 'data' in data:
-      self.write(data['data'])
-    self.finish()
-
-def on_connection_close(self):
-  '''
-  :return:
-  '''
-  # pylint: disable=attribute-defined-outside-init
-  self.connection_closed = True
+    chunk_size = int(4 * 1024 * 1024)
+    content_type = mimetypes.guess_type(path)
+    self.set_header("Content-Type", content_type[0])
+    with open(path, 'rb') as f:
+      while True:
+        chunk = f.read(chunk_size)
+        if not chunk:
+          break
+        try:
+          self.write(chunk)  # write the chunk to response
+          await self.flush() # send the chunk to client
+        except iostream.StreamCloseError:
+          # this means the client has closed the connection
+          # so break the loop
+          break
+        finally:
+          # deleting the chunk is very important because
+          # if many client are downloading files at the
+          # same time, the chunks in memory will keep
+          # increasing and will eat up the RAM
+          del chunk
+          # pause the coroutine so other handlers can run
+          await gen.sleep(0.000000001) # 1 nanosecond

heron/shell/src/python/handlers/filedatahandler.py

@@ -31,8 +31,7 @@ class FileDataHandler(tornado.web.RequestHandler):
   Responsible for reading and returning the file data given the offset
   and length of file to be read.
   """
-  @tornado.web.asynchronous
-  def get(self, path):
+  async def get(self, path):
     """ get method """
     if path is None:
       return {}
@@ -48,6 +47,5 @@ def get(self, path):
     if not os.path.isfile(path):
       return {}
     data = utils.read_chunk(path, offset=offset, length=length, escape_data=True)
-    self.write(json.dumps(data))
-    self.finish()
-    return None
+
+    await self.finish(json.dumps(data))

heron/shell/src/python/handlers/filehandler.py

@@ -30,20 +30,17 @@ class FileHandler(tornado.web.RequestHandler):
   Responsible for creating the web page for files. The html
   will in turn call the /filedata/ endpoint to get the file data.
   """
-  @tornado.web.asynchronous
-  def get(self, path):
+  async def get(self, path):
     """ get method """
     t = Template(utils.get_asset("file.html"))
     if path is None:
       self.set_status(404)
-      self.write("No such file")
-      self.finish()
+      await self.finish("No such file")
       return
 
     if not utils.check_path(path):
-      self.write("Only relative paths are allowed")
       self.set_status(403)
-      self.finish()
+      await self.finish("Only relative paths are allowed")
       return
 
     args = dict(
@@ -52,6 +49,5 @@ def get(self, path):
         pailer=utils.get_asset("jquery.pailer.js"),
         css=utils.get_asset("bootstrap.css"),
     )
-    self.write(t.generate(**args))
-    self.finish()
+    await self.finish(t.generate(**args))
     return

heron/shell/src/python/handlers/filestatshandler.py

@@ -31,8 +31,7 @@ class FileStatsHandler(tornado.web.RequestHandler):
   """
   Get the file stats in JSON format given the path.
   """
-  @tornado.web.asynchronous
-  def get(self, path):
+  async def get(self, path):
     ''' get method '''
     path = tornado.escape.url_unescape(path)
     if not path:
@@ -43,9 +42,8 @@ def get(self, path):
     # sandboxing. So we don't allow absolute paths and parent
     # accessing.
     if not utils.check_path(path):
-      self.write("Only relative paths are allowed")
       self.set_status(403)
-      self.finish()
+      await self.finish("Only relative paths are allowed")
       return
 
     listing = utils.get_listing(path)
@@ -69,5 +67,4 @@ def get(self, path):
             file_stats[fn]["path"] = tornado.escape.url_escape("/".join(path_fragments[:-1]))
       except:
         continue
-    self.write(json.dumps(file_stats))
-    self.finish()
+    await self.finish(json.dumps(file_stats))

heron/shell/src/python/handlers/healthhandler.py

@@ -27,9 +27,7 @@ class HealthHandler(tornado.web.RequestHandler):
   """
 
   # pylint: disable=attribute-defined-outside-init
-  @tornado.web.asynchronous
   def get(self):
     """ get method """
     self.content_type = 'plain/text'
-    self.write('ok')
-    self.finish()
+    self.finish('ok')

heron/shell/src/python/handlers/jmaphandler.py

@@ -31,13 +31,11 @@ class JmapHandler(tornado.web.RequestHandler):
   """
 
   # pylint: disable=attribute-defined-outside-init
-  @tornado.web.asynchronous
-  def get(self, pid):
+  async def get(self, pid):
     ''' get method '''
     utils.str_cmd(['rm', '-rf', '/tmp/heap.bin'], None, None)
     body = utils.str_cmd(['jmap', '-dump:format=b,file=/tmp/heap.bin',
                           str(pid)], None, None)
     utils.str_cmd(['chmod', '+r', '/tmp/heap.bin'], None, None)
     self.content_type = 'application/json'
-    self.write(json.dumps(body))
-    self.finish()
+    await self.finish(json.dumps(body))

heron/shell/src/python/handlers/jstackhandler.py

@@ -31,10 +31,8 @@ class JstackHandler(tornado.web.RequestHandler):
   """
 
   # pylint: disable=attribute-defined-outside-init
-  @tornado.web.asynchronous
-  def get(self, pid):
+  async def get(self, pid):
     ''' get method '''
     body = utils.str_cmd(['jstack', pid], None, None)
     self.content_type = 'application/json'
-    self.write(json.dumps(body))
-    self.finish()
+    await self.finish(json.dumps(body))

heron/shell/src/python/handlers/killexecutorhandler.py

@@ -32,8 +32,7 @@ class KillExecutorHandler(tornado.web.RequestHandler):
   """
   Responsible for killing heron-executor process.
   """
-  @tornado.web.asynchronous
-  def post(self):
+  async def post(self):
     """ post method """
     def status_finish(ret):
       self.set_status(ret)
@@ -69,7 +68,7 @@ def is_local():
           fh.close()
           logger.info("Killing process %s %s", instanceId, firstLine)
           os.kill(firstLine, signal.SIGTERM)
-          status_finish(200)
+          await status_finish(200)
       else: # instance_id not found
         logger.info("%s not found", filepath)
         status_finish(422)

heron/shell/src/python/handlers/memoryhistogramhandler.py

@@ -31,10 +31,8 @@ class MemoryHistogramHandler(tornado.web.RequestHandler):
   """
 
   # pylint: disable=attribute-defined-outside-init
-  @tornado.web.asynchronous
-  def get(self, pid):
+  async def get(self, pid):
     ''' get method '''
     body = utils.str_cmd(['jmap', '-histo', pid], None, None)
     self.content_type = 'application/json'
-    self.write(json.dumps(body))
-    self.finish()
+    await self.finish(json.dumps(body))

heron/shell/src/python/handlers/pidhandler.py

@@ -31,9 +31,7 @@ class PidHandler(tornado.web.RequestHandler):
   """
 
   # pylint: disable=attribute-defined-outside-init
-  @tornado.web.asynchronous
-  def get(self, instance_id):
+  async def get(self, instance_id):
     ''' get method '''
     self.content_type = 'application/json'
-    self.write(json.dumps(utils.chain([['cat', "%s.pid" % instance_id]])).strip())
-    self.finish()
+    await self.finish(json.dumps(utils.chain([['cat', "%s.pid" % instance_id]])).strip())

heron/shell/src/python/handlers/pmaphandler.py

@@ -31,10 +31,8 @@ class PmapHandler(tornado.web.RequestHandler):
   """
 
   # pylint: disable=attribute-defined-outside-init
-  @tornado.web.asynchronous
-  def get(self, pid):
+  async def get(self, pid):
     ''' get method '''
     body = utils.str_cmd(['pmap', '-pXX', pid], None, None)
     self.content_type = 'application/json'
-    self.write(json.dumps(body))
-    self.finish()
+    await self.finish(json.dumps(body))

heron/tools/tracker/src/python/routers/topologies.py

@@ -112,15 +112,14 @@ async def get_topology_config(
     role: Optional[str] = Query(None, deprecated=True),
 ):
   topology = state.tracker.get_topology(cluster, role, environ, topology)
-  topology_info = topology.info
-  return topology_info.physical_plan.config
+  return topology.info.physical_plan.config
 
 @router.get("/packingplan", response_model=TopologyInfoPackingPlan)
 async def get_topology_packing_plan(
-  cluster: str,
-  environ: str,
-  topology: str,
-  role: Optional[str] = Query(None, deprecated=True),
+    cluster: str,
+    environ: str,
+    topology: str,
+    role: Optional[str] = Query(None, deprecated=True),
 ):
   topology = state.tracker.get_topology(cluster, role, environ, topology)
   return topology.info.packing_plan
@@ -185,5 +184,4 @@ async def get_topology_logical_plan(
 
   """
   topology = state.tracker.get_topology(cluster, role, environ, topology)
-  topology_info = topology.info
-  return topology_info.logical_plan
+  return topology.info.logical_plan

integration_test/src/python/http_server/main.py

@@ -153,7 +153,7 @@ def main():
 
   logging.info("Starting server at port " + str(port))
   application.listen(port)
-  tornado.ioloop.IOLoop.instance().start()
+  tornado.ioloop.IOLoop.current().start()
 
 if __name__ == '__main__':
   main()