324 lines
12 KiB
Python
324 lines
12 KiB
Python
from tornado.httpclient import HTTPRequest, AsyncHTTPClient
|
|
from tornado.simple_httpclient import SimpleAsyncHTTPClient
|
|
import tornado.ioloop
|
|
import tornado.web
|
|
import os,sys,re,time
|
|
import threading
|
|
from tornado import gen
|
|
from functools import partial
|
|
import urllib.parse
|
|
import mimetypes
|
|
import math
|
|
import uuid
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import tornado.iostream
|
|
from tornado.escape import utf8
|
|
from tornado.log import gen_log
|
|
from ctypes import *
|
|
import struct
|
|
import os
|
|
#import time,datat
|
|
|
|
readchunky = False
|
|
total_downloaded = 0
|
|
threadpool = ThreadPoolExecutor(1) # A thread for reading chunks from the file
|
|
|
|
DEBUG = False
|
|
|
|
class NoQueueTimeoutHTTPClient(SimpleAsyncHTTPClient):
|
|
def fetch_impl(self, request, callback):
|
|
key = object()
|
|
|
|
self.queue.append((key, request, callback))
|
|
self.waiting[key] = (request, callback, None)
|
|
|
|
self._process_queue()
|
|
|
|
if self.queue:
|
|
gen_log.debug("max_clients limit reached, request queued. %d active, %d queued requests." % (len(self.active), len(self.queue)))
|
|
|
|
|
|
class Client:
|
|
def _gen_boundary(self, file_size):
|
|
if file_size < 1024:
|
|
blen = 10
|
|
else:
|
|
blen = math.ceil(math.log(file_size, 2))
|
|
bcnt = max(blen / 32, 1)
|
|
return "".join([str(uuid.uuid1()).replace("-", "") for _ in range(bcnt)])
|
|
print("_gen_boundary file_size",file_size)
|
|
def put_stream(self, url, pos, size, filename, on_response, chunk_size=1024):
|
|
'''Uploads file to provided url.
|
|
:param url: URL to PUT the file data to
|
|
:param filename: Name of the file (Content-Disposition header)
|
|
:param file_size: Size of the file in bytes. Used to produce a Content-Size header from file_size.
|
|
:param on_response: See AsyncHTTPClient.fetch
|
|
:return: Future content value under the provided url as binary string.
|
|
'''
|
|
uploadpos = pos
|
|
#file_size = os.path.getsize(filename) # Be aware: this could also block! In production, you need to do this in a separated thread too!
|
|
file_size = size # Be aware: this could also block! In production, you need to do this in a separated thread too!
|
|
|
|
print("filesize in put_stream is :",file_size)
|
|
ext = os.path.splitext(filename)[1].lower()
|
|
if ext:
|
|
content_type = mimetypes.types_map.get(ext, "application/octet-stream")
|
|
else:
|
|
content_type = "application/octet-stream"
|
|
|
|
enc_filename = urllib.parse.quote(filename)
|
|
# http://stackoverflow.com/questions/4526273/what-does-enctype-multipart-form-data-mean/28380690#28380690
|
|
boundary = self._gen_boundary(file_size)
|
|
CRLF = '\r\n'
|
|
post_head = b''.join(map(utf8, [
|
|
'--', boundary, CRLF,
|
|
# See https://tools.ietf.org/html/rfc5987 section 3.2.2 examples
|
|
'Content-Disposition: form-data; name="file"; filename*=utf-8\'\'%s' % enc_filename, CRLF,
|
|
'Content-Type: ', content_type, CRLF,
|
|
'Content-Transfer-Encoding: binary', CRLF,
|
|
CRLF,
|
|
]))
|
|
post_tail = b''.join(map(utf8, [
|
|
CRLF, '--', boundary, '--', CRLF
|
|
]))
|
|
content_length = len(post_head) + int(file_size) + len(post_tail)
|
|
#print("content_length is:",content_length)
|
|
headers = {
|
|
'Content-Type': 'multipart/form-data; boundary=' + boundary,
|
|
'Content-Transfer-Encoding': 'binary',
|
|
'Content-Length': str(content_length),
|
|
}
|
|
|
|
@gen.coroutine
|
|
def body_producer(write):
|
|
if DEBUG:
|
|
sys.stdout.write(post_head.decode('ascii'))
|
|
write(post_head)
|
|
remaining = file_size
|
|
#print("remaining in body_produceris :",remaining)
|
|
with open(filename, "rb") as fileobj:
|
|
fileobj.seek(int(uploadpos))
|
|
while remaining > 0:
|
|
#print("uploadpos in while is :",int(uploadpos))
|
|
# data = yield threadpool.submit(fileobj.read(int(remaining)), chunk_size)
|
|
data = fileobj.read(int(remaining))
|
|
# print(str(data,encoding = "utf-8"))
|
|
if data:
|
|
remaining -= len(data)
|
|
print("len(data) in if is :",len(data),"uploadpos in while is :",int(uploadpos),"remaining in if is :",len(data))
|
|
#print("data in if is :",data)
|
|
#print("remaining in if is :",len(data))
|
|
if DEBUG:
|
|
sys.stdout.write(data.decode('utf-8'))
|
|
yield write(data)
|
|
else:
|
|
break
|
|
if DEBUG:
|
|
sys.stdout.write(post_tail.decode('ascii'))
|
|
write(post_tail)
|
|
|
|
request = tornado.httpclient.HTTPRequest(url=url, request_timeout=1200, method='POST', headers=headers,body_producer=body_producer)
|
|
print("before1:"+threading.current_thread().getName()+url)
|
|
|
|
return tornado.httpclient.AsyncHTTPClient().fetch(request, callback=on_response)
|
|
|
|
|
|
def geturlread(action,host,filepath,uid,gid,pos,size):
|
|
if action=="read":
|
|
url = "http://"+host+"/read?filepath="+filepath+"&uid="+uid+"&gid="+gid+"&pos="+str(pos)+"&size="+str(size)
|
|
print(url)
|
|
return url
|
|
|
|
def geturlupload(action,host,targetpath,pos,size,totalsize):
|
|
if action=="upload":
|
|
url = "http://"+host+"/upload?targetpath="+targetpath+"&pos="+str(pos)+"&size="+str(size)+"&totalsize="+str(totalsize)
|
|
print(url)
|
|
return url
|
|
|
|
'''def chunky(path,pos,totalsize,chunk):
|
|
#print("self._max_body_size",self._max_body_size)
|
|
print("chunk length",len(chunk))
|
|
global total_downloaded
|
|
global readchunky
|
|
if readchunky == False and os.path.exists(path):
|
|
os.remove(path)
|
|
readchunky = True
|
|
if not os.path.exists(path):
|
|
f = open(path,'w')
|
|
f.close()
|
|
f = open(path,'rb+')
|
|
chunklength = len(chunk)-8
|
|
pos,chunknew = struct.unpack('l%ds'%chunklength,chunk)
|
|
print("decode pos is",pos)
|
|
f.seek(pos)
|
|
f.write(chunknew)
|
|
print("after write position,chunksize",f.tell(),len(chunknew))
|
|
pos = pos+len(chunknew)
|
|
#f.flush()
|
|
if pos==totalsize:
|
|
f.close()
|
|
total_downloaded += len(chunknew)'''
|
|
# the OS blocks on file reads + writes -- beware how big the chunks is as it could effect things
|
|
|
|
def sizebwchunky(chunk):
|
|
global FILESIZE
|
|
FILESIZE = int(chunk)
|
|
|
|
@gen.coroutine
|
|
def sizebw(host,filepath):
|
|
url = "http://"+host+"/sizebw?filepath="+filepath
|
|
print(url)
|
|
request = HTTPRequest(url, streaming_callback=partial(sizebwchunky), request_timeout=300)
|
|
AsyncHTTPClient.configure('tornado.simple_httpclient.SimpleAsyncHTTPClient', max_body_size=1024*1024*1024)
|
|
http_client = AsyncHTTPClient(force_instance=True)
|
|
#AsyncHTTPClient.configure("tornado.curl_httpclient.CurlAsyncHTTPClient")
|
|
#http_client = AsyncHTTPClient()
|
|
response = yield http_client.fetch(request)
|
|
tornado.ioloop.IOLoop.instance().stop()
|
|
|
|
@gen.coroutine
|
|
def writer(host,filepath,targetdir,uid,gid,pos,size):
|
|
file_name = targetdir
|
|
path = file_name
|
|
chunk_size = 80*1024*1024
|
|
no = int(size) // chunk_size
|
|
i = 0
|
|
global total_downloaded
|
|
global readchunky
|
|
lib=cdll.LoadLibrary('./libpycall.so')
|
|
func=lib.update_bitmap
|
|
func.argtypes=(c_int,c_int,c_char_p,c_int)
|
|
#if readchunky == False and os.path.exists(path):
|
|
# os.remove(path)
|
|
# readchunky = True
|
|
#if not os.path.exists(path):
|
|
# f = open(path,'w')
|
|
# f.close()
|
|
while i<no:
|
|
request = HTTPRequest(geturlread("read",host,filepath,uid,gid,pos,str(chunk_size)),request_timeout=300)
|
|
pos = str(int(pos)+chunk_size)
|
|
i = i+1
|
|
http_client = AsyncHTTPClient()
|
|
response = yield http_client.fetch(request)
|
|
response = response.body
|
|
print("chunk length",len(response))
|
|
f = open(path,'rb+')
|
|
chunklength = len(response)-8
|
|
posnew,chunknew = struct.unpack('l%ds'%chunklength,response)
|
|
print("posnew is",posnew)
|
|
print("chunkleng",len(chunknew))
|
|
print("filesize",FILESIZE)
|
|
f.seek(posnew)
|
|
f.write(chunknew)
|
|
total_downloaded=total_downloaded+len(chunknew)
|
|
#print("after write position,chunksize",f.tell(),len(chunknew))
|
|
f.close()
|
|
func= lib.update_bitmap(int(posnew),int(len(chunknew)),filepath.encode("utf-8"),FILESIZE)
|
|
print ("finish")
|
|
print("total bytes downloaded was", total_downloaded)
|
|
if (int(size) % chunk_size) != 0:
|
|
last = int(size) % chunk_size
|
|
request = HTTPRequest(geturlread("read",host,filepath,uid,gid,pos,str(last)),request_timeout=300)
|
|
http_client = AsyncHTTPClient()
|
|
response = yield http_client.fetch(request)
|
|
response = response.body
|
|
#print("chunk length",len(response))
|
|
f = open(path,'rb+')
|
|
chunklength = len(response)-8
|
|
posnew,chunknew = struct.unpack('l%ds'%chunklength,response)
|
|
#print("posnew is",posnew)
|
|
f.seek(posnew)
|
|
f.write(chunknew)
|
|
total_downloaded=total_downloaded+len(chunknew)
|
|
#print("after write position,chunksize",f.tell(),len(chunknew))
|
|
f.close()
|
|
func= lib.update_bitmap(int(posnew),int(len(chunknew)),filepath.encode("utf-8"),FILESIZE)
|
|
print ("finish")
|
|
print("total bytes downloaded was", total_downloaded)
|
|
if total_downloaded==realsize:
|
|
tornado.ioloop.IOLoop.instance().stop()
|
|
''' lib=cdll.LoadLibrary('./libpycall.so')
|
|
print("start")
|
|
func=lib.update_bitmap
|
|
func.argtypes=(c_int,c_int,c_char_p,c_int)
|
|
func=lib.update_bitmap(int(pos),int(size),filepath.encode("utf-8"),FILESIZE)
|
|
print("finish")
|
|
'''
|
|
|
|
|
|
@gen.coroutine
|
|
def upload(host,filepath,targetpath,pos,size):
|
|
def on_response(request):
|
|
print("on_response:"+threading.current_thread().getName())
|
|
|
|
print("=============== GOT RESPONSE ======")
|
|
print(request.body.decode('ascii'))
|
|
#print(request.body)
|
|
print("===================================")
|
|
tornado.ioloop.IOLoop.current().stop() # Stop the loop when the upload is complete.
|
|
|
|
client = Client()
|
|
print("before"+threading.current_thread().getName())
|
|
#tornado.ioloop.IOLoop.instance().start()
|
|
yield client.put_stream(geturlupload("upload",host,targetpath,pos,size,os.path.getsize(filepath)), pos, size, filepath, on_response)
|
|
print("after:"+threading.current_thread().getName())
|
|
#tornado.ioloop.IOLoop.current().stop() # Stop the loop when the upload is complete.
|
|
|
|
|
|
|
|
def readentrance(host,filepath,targetdir,uid,gid,pos,size):
|
|
print(host,filepath,targetdir,uid,gid,pos,size)
|
|
sizebw(host,filepath)
|
|
tornado.ioloop.IOLoop.instance().start()
|
|
filesize = FILESIZE
|
|
streamno = 3
|
|
start_time = time.time()
|
|
global realsize
|
|
if(int(size)>=filesize):
|
|
realsize = filesize
|
|
streamsize = (filesize-int(pos)) // (streamno-1)
|
|
else:
|
|
realsize = int(size)
|
|
streamsize = (int(size)) // (streamno-1)
|
|
i = 0
|
|
threads = []
|
|
while i < (streamno-1):
|
|
threads.append(threading.Thread(target=writer,args=(host,filepath,targetdir,uid,gid,int(pos)+streamsize*i,streamsize)))
|
|
i=i+1
|
|
if (streamsize*i) < realsize:
|
|
threads.append(threading.Thread(target=writer,args=(host,filepath,targetdir,uid,gid,int(pos)+streamsize*i,realsize-streamsize*i)))
|
|
for t in threads:
|
|
t.setDaemon(True)
|
|
print("thread name is :",t.getName())
|
|
t.start()
|
|
tornado.ioloop.IOLoop.instance().start()
|
|
for t in threads:
|
|
t.join()
|
|
end_time = time.time()
|
|
print("Total time :{}".format(end_time-start_time))
|
|
|
|
def uploadentrance(host,filepath,targetpath):
|
|
streamno = 3
|
|
start_time = time.time()
|
|
filesize = os.path.getsize(filepath)
|
|
streamsize = filesize // (streamno-1)
|
|
i = 0
|
|
threads = []
|
|
while i < (streamno-1):
|
|
threads.append(threading.Thread(target=upload,args=(host,filepath,targetpath,streamsize*i,streamsize)))
|
|
i=i+1
|
|
if (streamsize*i) < filesize:
|
|
threads.append(threading.Thread(target=upload,args=(host,filepath,targetpath,streamsize*i,filesize-streamsize*i)))
|
|
|
|
for t in threads:
|
|
t.setDaemon(True)
|
|
print("thread name is :",t.getName())
|
|
t.start()
|
|
tornado.ioloop.IOLoop.instance().start()
|
|
#for t in threads:
|
|
t.join()
|
|
end_time = time.time()
|
|
print("Total time :{}".format(end_time-start_time))
|
|
#tornado.ioloop.IOLoop.current().stop() # Stop the loop when the upload is complete.
|
|
|