代码拉取完成,页面将自动刷新
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.
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。