python練習七—P2P下載

最近有些事兒比較忙，python的學習就斷斷續續，這個練習來得比預期的晚，不過還好，不管做什么，我都希望能認真對待，認真做好每一件事。

引入

這個練習原書中稱作“使用XML-RPC進行文件共享”，題目是從使用的技術介紹的，做完這個練習之后我覺得最終其實實現的是一個迅雷的雛形——一個P2P下載器。當然了實際使用中的下載工具遠遠比這個要復雜，包括斷點續傳，斷網重連等等。

那么我們學習的方法也就明確了，使用到了P2P的概念和XML-RPC的原理，那么就從這兩個開始吧。

 

關於P2P

額，這里不進行官方的概念介紹，結合這個練習說明P2P的大體原理。使用P2P之后，我們查找一個共享資源的大體流程如下

# P2P協議的理解，弱化了server-client的概念，在網絡中，每個節點都可能是server，也可能是client，暫且把發起請求的稱為client，處理請求的稱為server

1. client向服務器A發起尋找文件test的請求
2. A現在自己的目錄下面尋找是否有test
3. 如果有則返回該文件給client，程序結束
4. 如果A服務器沒有，則A遍歷在線節點列表
5. 如果找到，返回給client，結束
6. 如果沒有，返回空

由上的過程看出網絡中每個節點需要具備以下功能：

# 每個節點需要維護的資源
共享目錄：在網絡中存在n台機器，每台服務器都有一個可供其他服務器訪問的目錄（目錄下面有各種文件）
在線節點列表：每天機器都有一個所有願意共項目錄的服務器列表

# 服務器的功能
服務器應該能更新在線節點列表
如果在線節點很多，是不是應該一直尋找下去（設置一個查找深度）
服務器查詢某個文件是否存在
服務器可以下載某一個文件到自己的共享目錄

很明白了，在P2P網絡中每個節點可以為別人提供資源，也可以向其他節點請求資源，對於每個節點來說就是共享出自己的資源，然后知道去哪里請求資源（需要一個其他節點的列表）。

關於xml-rpc

rpc：遠程過程調用（之前學習分布式操作系統的時候想不到遠程過程調用是什么，使用場景是什么，但是相信總起到了我意識到或者沒意識到的作用），心愛那個心愛那個其實和http協議也有類似的地方，都是請求其他計算機的資源，並不關心在其他計算機上是怎么實現的。其原理也就是

# xml_rpc原理
在server A定義方法method
在client使用A的proxy調用method

當然了，rpc還有更詳細的協議規范，但是這一層都被xml-rpc通過封裝屏蔽了。這里也不作進一步探尋了。

具體實現

主要有兩個模塊，client和server，主要功能在server中實現，client只是對server調用進行了一些封裝。

server.py

#!/usr/bin/env python
# -*- coding=utf-8 -×-

from xmlrpclib import ServerProxy, Fault
from os.path import join, isfile, abspath
from SimpleXMLRPCServer import SimpleXMLRPCServer
from urlparse import urlparse
import sys

MAX_HISTORY_LENGTH = 6
SimpleXMLRPCServer.allow_reuse_address = 1

UNHANDLED = 100
ACCESS_DENIED = 200

class UnhandledQuery(Fault):
    """
    表示無法處理的查詢異常
    """
    def __init__(self, message='Could not handle the query'):
        Fault.__init__(self, UNHANDLED, message)


class AccessDenied(Fault):
    """
    在用戶試圖訪問未被授權的資源時引發的異常
    """
    def __init__(self, message='Access denied'):
        Fault.__init__(self, ACCESS_DENIED, message)

def inside(dir, name):
    """
    判斷訪問的目錄是否有限制訪問的目錄，限制非法目錄訪問，比如/var/www/../data
    """
    dirs = abspath(dirs)
    name = abspath(name)
    return name.startswith(join(dirs, ''))



def getPort(url):
    """
    根據url獲取端口號
    """
    name = urlparse(url)[1]
    parts = name.split(":")
    return int(parts[-1])

class Node:
    """
    P2P網絡中的節點
    """
    def __init__(self, url, directory, secret):
        self.url = url
        self.directory = directory
        self.secret = secret
        self.know = set()

    def query(self, query, history=[]):
        """
        查找文件，先在本機查找，如果找到則返回，找不到則查找其他已知節點
        """
        try:
            print 'search local'
            return self._handle(query)
        except UnhandledQuery:
            # 添加到history，標記查找的深度
            history = history + [self.url]
            if len(history) >= MAX_HISTORY_LENGTH:
                raise
            print 'search other server'
            return self._broadcast(query, history)

    def hello(self, other):
        """
        認識其他節點
        """
        self.know.add(other)
        return 0
    def fetch(self, query, secret):
        """
        用於從節點下載數據
        """
        print 'server.fetch'
        if secret != self.secret:
            raise AccessDenied
        result = self.query(query)
        print 'result----',result
        # 把查詢到的數據寫到本地
        f = open(join(self.directory, query), 'w')
        f.write(result)
        f.close()
        return 0

    def _start(self):
        """
        內部使用，用於啟動XML_RPC服務器
        """
        server = SimpleXMLRPCServer(("", getPort(self.url)), logRequests=False)
        server.register_instance(self)
        server.serve_forever()

    def _handle(self, query):
        """
        搜索文件在本服務器上是否存在
        """
        dirs = self.directory
        name = join(dirs, query)
        if not isfile(name):
            raise UnhandledQuery
        return open(name).read()

    def _broadcast(self, query, history):
        """
        內部時使用，用於將查詢廣播到其他已知節點
        """
        for other in self.know.copy():
            try:
                # 根據url創建遠程節點的proxy，使用xml_rpc進行遠程調用
                print 'search ----', other
                server = ServerProxy(other)
                print 'start query', other
                return server.query(query, history)
            except Fault, f:
                if f.faultCode == UNHANDLED:
                    pass
                else:
                    self.know.remove(other)
            except:
                # 說明該server已經不可用
                self.know.remove(other)
        # 如果在所有節點中沒有找到，就返回空
        raise UnhandledQuery


def main():
    url, directory, secret = sys.argv[1:]
    node = Node(url, directory, secret)
    node._start()

if __name__ == '__main__':
    main()

 

client.py

#! /usr/bin/env python
# -*- coding=utf-8 -*-

from xmlrpclib import ServerProxy, Fault
from cmd import Cmd
from random import choice
from string import lowercase
from server import Node, UNHANDLED
from threading import Thread
from time import sleep
import sys

HEAD_START = 0.1 # second
SECRET_LEN = 100

def randomStr(len):
    """
    生成指定長度的字符串
    """
    chars = []
    letters = lowercase[:26]
    while len > 0:
        len = len - 1
        chars.append(choice(letters))
    return ''.join(chars)

class Client1(Cmd):
    """
    Node基於文本的界面
    """

    def __init__(self, url, dirname, urlfile):
        """
        初始化Cmd
        隨機生成秘鑰
        啟動服務器
        啟動字符界面
        """
        print 'in'
        Cmd.__init__(self)
        self.secret = randomStr(SECRET_LEN)
        n = Node(url, dirname, self.secret)
        t = Thread(target=n._start)
        t.setDaemon(1)
        t.start()
        # 等待服務器啟動
        sleep(HEAD_START)
        self.server = ServerProxy(url)
        for line in open(urlfile):
            line = line.strip()
            self.server.hello(line)

    def do_fetch(self, arg):
        "調用服務器的fetch方法"
        try:
            print 'do_fetch'
            self.server.fetch(arg, self.secret)
        except Fault, f:
            print f
            if f.faultCode != UNHANDLED:
                raise
            print 'Could not find the file ', arg

    def do_exit(self, arg):
        "退出程序"
        print
        sys.exit()

    do_EOF = do_exit # 接收到EOF的時候也退出程序
    # 設置提示符
    prompt = '>'

def main():
    urlfile, directory, url = sys.argv[1:]
    print urlfile, directory, url
    print '---------'
    print dir()
    client = Client1(url, directory, urlfile)
    client.cmdloop()

if __name__ == '__main__':
    main()

 

總結

在運行這個練習的時候，要注意死鎖，因為有共享資源的問題

# 線程死鎖
server A的url list里面有Server B
server B的url list里面有Server A
當A請求B的時候，B發現自己也沒有，B根據url list查詢其他server，這個時候根據list需要去查詢A
但是這個時候A正在等待B的回應，所以就造成了
A等B，B等A的死鎖情況

# 解決死鎖
每個server B在向url list里面的server A發出請求之前判斷本神是否是Server A正在請求B

 

---

 

 

完整代碼

http://pan.baidu.com/s/1i58mk3V