Skip to content

T
tcp-trans
Commit f499ec6e by ziho
Options

交换了提出连接请求的主机

parent 2615eb14
Showing with 347 additions and 89 deletions
oldver-接收方有公网/tcprecv.py 0 → 100644
import os
import os
import socket
import threading
import time
dst='recv'
tempdir='recvTempFile'
port=43215
delete_command='del'+' ' #'del ' for windows, 'rm ' for linux
def handle_file(sock):
data = b''
while True:
part = sock.recv(1024)
data += part
if len(part) < 1024:
break
filename = data.decode('utf-8')
sock.sendall('ACK'.encode('utf-8'))
with open(os.path.join(tempdir,filename), 'wb') as F:
part = sock.recv(4096)
while part:
F.write(part)
part = sock.recv(4096)
comm='tar -xf '+os.path.join(tempdir,filename)+' -C '+dst
os.system(comm)
comm=delete_command+os.path.join(tempdir,filename)
os.system(comm)
print(filename+'...received!')
if not os.path.exists(dst):
os.system('mkdir '+dst)
print('[info] create dir:'+dst)
if not os.path.exists(tempdir):
os.system('mkdir '+tempdir)
print('[info] create dir:'+tempdir)
runMainLoop=True
def listenforkeyboard():
while True:
aaa=input('[info] input <quit/exit> can quit:\n')
print('you input:'+aaa)
if aaa=='quit' or aaa=='exit':
global runMainLoop
runMainLoop=False
break
a=''
while not (a=='Y' or a=='y'):
a=input('start receiving?<Y or y> :')
t = threading.Thread(target=listenforkeyboard)
t.start()
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', port))
server.listen(10)
server.settimeout(1)
while runMainLoop:
try:
sock, _ = server.accept()
except:
pass
else:
n=threading.Thread(target=handle_file, args=(sock,))
n.start()
print('[info] stop and exit! will delete temp file (or not)')
print('[info] and all the thread will end (dont know what will happen)')
\ No newline at end of file
oldver-接收方有公网/tcpsend.py 0 → 100644
import re
import re
import os
import socket
import threading
import time
ip='218.104.194.130'
port=43215
source='selfplay_data'
packsize=64 #the number of files that 1 '*.tar' file contain
maxthread=5
sleeptime=3 #I will try to change it dynamically
alpha=0.8 #maybe moving average?
#if there are many threads sending at the same time,they may be slow together
delete_command='rm'+' ' #'del ' for windows, 'rm ' for linux
packlist=[]
packindex=0
tempdir='sendTempFile'
if not os.path.exists(tempdir):
os.system('mkdir '+tempdir)
print('[info] create dir:'+tempdir)
def getPrefix(path):
'''
get prefix
'''
filelist=os.listdir(source)
prefix={}
for f in filelist:
matchResult=re.match(r'(agz9_selfplay_zhuque_([-A-Za-z1-9]+)_mlu\d_)(\d+)\.moves',f)
if not matchResult:
continue
pre=matchResult.group(1)
index=int(matchResult.group(3))
if pre in prefix:
prefix[pre]=min(prefix[pre],index)
else:
prefix[pre]=index
return prefix
def sendFileThread(ip, port,packlist,packindex):
packname='tempPack'+str(packindex)+'.tar'
out='sending '+packname+'...'
comm='tar -C '+source+' -cf '+os.path.join(tempdir,packname)
for f in packlist:
comm+=' '+f
os.system(comm)
while True:
time_start=time.time()
try:
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect((ip, port))
client.send(packname.encode())
respond = client.recv(10)
if respond == b'ACK':
with open(os.path.join(tempdir,packname), 'rb') as f:
while True:
part = f.read(2048)
if not part:
break
client.sendall(part)
except Exception as e:
print(e)
print(out+'retry')
else:
client.close()
time_end=time.time()
global sleeptime
print(out+'done!total time={:.3f} sec,sleep time={:.3f} sec'.format(time_end-time_start,sleeptime))
# maybe I can use conn time to change sleeptime dynamically?
sleeptime=min(5,alpha*sleeptime+(1-alpha)*(time_end-time_start)*1.375)
os.system(delete_command+os.path.join(tempdir,packname))
break
runMainLoop=True
def listenforkeyboard():
while True:
aaa=input('[info] input <quit/exit> can quit:\n')
print('you input:'+aaa)
if aaa=='quit' or aaa=='exit':
global runMainLoop
runMainLoop=False
break
a=''
while not (a=='Y' or a=='y'):
prefix=getPrefix(source)
print('{} prefixes had found'.format(len(prefix)))
a=input('start sending?<Y or y> :')
t = threading.Thread(target=listenforkeyboard)
t.start()
while runMainLoop:
for key in prefix:
if len(threading.enumerate()) >= maxthread:
continue
path=os.path.join(source,key+str(prefix[key])+'.moves.finish')
if os.path.exists(path):
filename=key+str(prefix[key])+'.moves'
packlist.append(filename)
if len(packlist)==packsize:
if packindex>1000:
packindex=0
t = threading.Thread(target=sendFileThread, args=(ip, port,packlist,packindex))
t.setDaemon(True) #thread will end when main thread end
t.start()
packlist=[]
packindex+=1
time.sleep(sleeptime) #wait for the socket connected
prefix[key]+=1
print('[info] stop and exit! will delete temp file (or not)!')
print('[info] and all the thread will end (dont know what will happen)')
readme.md
# 修改中，勿使用
# 【修改中】127.0.0.1已完成，外网未测试
## tcp传输
......@@ -7,9 +7,16 @@
**1、接收文件的一端,**需打开文件**tcprecv.py**修改参数
``` python
ip='127.0.0.1' #目标主机IP
dst='recv' #文件存放地点，会把.moves文件送进去
tempdir='recvTempFile' #临时文件
port=43215 #随意指定的端口号，没有冲突就不管
maxthread=5 #最大开启的线程数量
sleeptime=3 #请求文件的最少间隔，因为同一时刻有多个线程大家会一起变慢
alpha=0.8 #尝试用滑动平均法来控制sleeptime,网络慢的时候就发慢些
#alpha是上一时刻sleeptime的权重
```
**2、发送文件的一端**,需打开**tcpsend.py**文件修改参数，必须要改的是IP
......@@ -17,30 +24,20 @@ port=43215 #随意指定的端口号，没有冲突就不管
​
``` python
ip='127.0.0.1' #目标主机IP
port=43215 #随意指定的端口号，没有冲突就不管
source='selfplay_data' #存放有.moves和.moves.finish的地方
packsize=64 #打包时多少个文件一组，小批次的话文件到达得就均匀些，文件产生慢的话似乎也没必要
maxthread=5 #最大开启的线程数量
sleeptime=3 #每两个打包文件发送的最少间隔，因为同一时刻有多个线程大家会一起变慢
alpha=0.8 #尝试用滑动平均法来控制sleeptime,网络慢的时候就发慢些
#alpha是上一时刻sleeptime的权重
```
**3、开启接受端**
tempdir='sendTempFile' #临时位置
waitinglist=[] #这个是还未发送的文件队列，接收到请求时，取出一个进入发送文件的线程
maxExistTempFile=25 #专门有线程负责打包文件，这个是最大打包文件个数，如果打包速度跟不上传输速度，就可能需要把这个数稍微调大
```
python tcprecv.py
```
需要输入y确认启动，（不启动就ctrl-c），运行中输入quit或exit可以退出
**4、开启发送端**
**3、（在有.moves和.moves.finish的，而且有公网IP的服务器上）开启发送端**
```bash
......@@ -57,6 +54,17 @@ python tcpsend.py
**4、（在没有外网那台服务器）开启接受端**
```
python tcprecv.py
```
需要输入y确认启动，（不启动就ctrl-c），运行中输入quit或exit可以退出
#### 实现
......@@ -76,3 +84,7 @@ $$
​ 网络波动大的时候情况还没测试，因为只测试过127.0.0.1的情况
​ 也许排队论有一些确定方法
2、变成请求文件的模式多了一些步骤，可能还有没考虑的情况
\ No newline at end of file
tcprecv.py
......@@ -3,41 +3,63 @@ import socket
import threading
import time
ip='120.236.247.203'
dst='recv'
tempdir='recvTempFile'
port=43215
maxthread=5
sleeptime=1
alpha=0.8
delete_command='rm'+' ' #'del ' for windows, 'rm ' for linux
rbuf=512*1024
def handle_file(sock):
data = b''
if not os.path.exists(dst):
os.system('mkdir '+dst)
print('[info] create dir:'+dst)
if not os.path.exists(tempdir):
os.system('mkdir '+tempdir)
print('[info] create dir:'+tempdir)
def unzip(packname):
comm='tar -xf '+os.path.join(tempdir,packname)+' -C '+dst
os.system(comm)
comm=delete_command+os.path.join(tempdir,packname)
os.system(comm)
def requestThread():
global sleeptime
while True:
part = sock.recv(1024)
data += part
if len(part) < 1024:
time_start=time.time()
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET,socket.SO_RCVBUF,rbuf)
sock.connect((ip, port))
data = sock.recv(50)
filename = data.decode()
if filename=='wait':
sleeptime=5
break
filename = data.decode('utf-8')
sock.sendall('ACK'.encode('utf-8'))
sock.sendall('ACK'.encode())
with open(os.path.join(tempdir,filename), 'wb') as F:
part = sock.recv(4096)
part = sock.recv(65536)
while part:
F.write(part)
part = sock.recv(4096)
comm='tar -xf '+os.path.join(tempdir,filename)+' -C '+dst
os.system(comm)
comm=delete_command+os.path.join(tempdir,filename)
os.system(comm)
print(filename+'...received!')
part = sock.recv(65536)
except Exception as e:
print(e)
print('request failed...retry')
else:
#sock.close()
time_end=time.time()
print('request '+filename+'...done!total time={:.3f} sec,sleep time={:.3f} sec'.format(time_end-time_start,sleeptime))
sleeptime=min(5,alpha*sleeptime+(1-alpha)*(time_end-time_start)*1.375)
if not os.path.exists(dst):
os.system('mkdir '+dst)
print('[info] create dir:'+dst)
if not os.path.exists(tempdir):
os.system('mkdir '+tempdir)
print('[info] create dir:'+tempdir)
unzip(filename)
break
runMainLoop=True
def listenforkeyboard():
......@@ -56,19 +78,16 @@ while not (a=='Y' or a=='y'):
t = threading.Thread(target=listenforkeyboard)
t.start()
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', port))
server.listen(10)
server.settimeout(1)
while runMainLoop:
try:
sock, _ = server.accept()
except:
pass
else:
n=threading.Thread(target=handle_file, args=(sock,))
n.start()
if len(threading.enumerate())<maxthread:
t=threading.Thread(target=requestThread)
t.setDaemon(True)
t.start()
time.sleep(sleeptime)
print('[info] stop and exit! will delete temp file (or not)')
print('[info] and all the thread will end (dont know what will happen)')
\ No newline at end of file
tcpsend.py
......@@ -4,20 +4,18 @@ import socket
import threading
import time
ip='218.104.194.130'
port=43215
source='selfplay_data'
packsize=64 #the number of files that 1 '*.tar' file contain
maxthread=5
sleeptime=3 #I will try to change it dynamically
alpha=0.8 #maybe moving average?
#if there are many threads sending at the same time,they may be slow together
delete_command='rm'+' ' #'del ' for windows, 'rm ' for linux
packlist=[]
packindex=0
packindex=0 #index for tar
maxindex=10000
tempdir='sendTempFile'
waitinglist=[]
maxExistTempFile=25
delete_command='rm'+' ' #'del ' for windows, 'rm ' for linux
if not os.path.exists(tempdir):
os.system('mkdir '+tempdir)
......@@ -42,39 +40,62 @@ def getPrefix(path):
prefix[pre]=index
return prefix
def sendFileThread(ip, port,packlist,packindex):
def tarFileThread():
global packindex
global prefix
global waitinglist
packlist=[]
while True:
for key in prefix:
if len(waitinglist)>maxExistTempFile:
continue
path=os.path.join(source,key+str(prefix[key])+'.moves.finish')
if os.path.exists(path):
filename=key+str(prefix[key])+'.moves'
packlist.append(filename)
if len(packlist)==packsize:
packname='tempPack'+str(packindex)+'.tar'
out='sending '+packname+'...'
comm='tar -C '+source+' -cf '+os.path.join(tempdir,packname)
for f in packlist:
comm+=' '+f
os.system(comm)
waitinglist.append(packname)
packlist=[]
packindex+=1
if packindex>maxindex:
packindex=0
prefix[key]+=1
def sendThread(sock,packname):
global waitinglist
while True:
time_start=time.time()
try:
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect((ip, port))
client.send(packname.encode('utf-8'))
respond = client.recv(5)
if not packname:
print('no zip file to send yet!')
sock.send('wait'.encode())
break
out='sending '+packname+'...'
sock.send(packname.encode())
respond = sock.recv(10)
if respond == b'ACK':
with open(os.path.join(tempdir,packname), 'rb') as f:
while True:
part = f.read(2048)
part = f.read(8192)
if not part:
break
client.sendall(part)
sock.sendall(part)
sock.close()
except Exception as e:
print(e)
print(out+'retry')
sock.close()
waitinglist.append(packname)
break
else:
client.close()
time_end=time.time()
global sleeptime
print(out+'done!total time={:.3f} sec,sleep time={:.3f} sec'.format(time_end-time_start,sleeptime))
# maybe I can use conn time to change sleeptime dynamically?
sleeptime=min(5,alpha*sleeptime+(1-alpha)*(time_end-time_start)*1.375)
print(out+'done!used time={:.3f} sec'.format(time_end-time_start))
os.system(delete_command+os.path.join(tempdir,packname))
break
......@@ -97,25 +118,31 @@ while not (a=='Y' or a=='y'):
t = threading.Thread(target=listenforkeyboard)
t.start()
while runMainLoop:
for key in prefix:
if len(threading.enumerate()) >= maxthread:
continue
t = threading.Thread(target=tarFileThread)
t.setDaemon(True)
t.start()
path=os.path.join(source,key+str(prefix[key])+'.moves.finish')
if os.path.exists(path):
filename=key+str(prefix[key])+'.moves'
packlist.append(filename)
if len(packlist)==packsize:
if packindex>1000:
packindex=0
t = threading.Thread(target=sendFileThread, args=(ip, port,packlist,packindex))
t.setDaemon(True) #thread will end when main thread end
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', port))
server.listen(5)
server.settimeout(1)
while runMainLoop:
try:
sock, _ = server.accept()
except:
pass
else:
if len(waitinglist):
packname=waitinglist.pop(0)
else:
packname=None
t=threading.Thread(target=sendThread,args=(sock,packname))
t.setDaemon(True)
t.start()
packlist=[]
packindex+=1
time.sleep(sleeptime) #wait for the socket connected
prefix[key]+=1
print('[info] stop and exit! will delete temp file (or not)!')
print('[info] and all the thread will end (dont know what will happen)')
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment