并发编程练习

 计算机操作系统的中断的作用：cpu会切：io阻塞，程序运行时间过长
 
    中断：计算机执行期间，系统内发生任何非寻常的或非预期的急需处理事件，使得
cpu暂时中断当前正在执行的程序而转去执行相应的事件处理程序。
     
    待处理完毕后又返回原来被中断处理急需执行或者调度新的进程执行的过程，它使计算
机可以更好更快的利用有限的系统资源解决系统响应速度和运行效率的一种控制技术：
    实时响应 + 系统调用
     
    中断装置是由一些特定的寄存器和控制线路组成，中央处理器和外围设备等识别到的
事件保存在特定的寄存器中。
    中央处理器每执行完一条指令，均有中断装置判别是否有事件发生。
    若无事件发生，CPU继续执行。
    若有事件发生，则中断装置中断原占有CPU的程序的执行，让操作系统的处理事件服
务程序占用CPU，对出现的事件进行处理，事件处理完后，再让原来的程序继续占用CPU执行

   操作系统的核心是内核，独立于普通的应用程序，内核可以访问受保护的内存空间，
也可以访问底层硬件设备的所有权限，为了保证用户进程不能直接操作内核，保证内核
的安全，操作系统将虚拟空间划分为两部分，一部分是内核空间，一部分是用户空间。
 
    内核态：运行操作系统的程序，os的数据存放
     
    用户态：运行用户程序，用户进程的数据存放
 
    用户态的应用程序可以通过三种方式来访问内核态的资源：
        1）系统调用
        2）库函数
        3）Shell脚本
    用户态到内核态的切换:
        1.系统调用        用户程序主动发起的 软中断 os.fork() process
        2.异常            被动的   当CPU正在执行运行在用户态的程序时，突然发生某些预
先不可知的异常事件，这个时候就会触发从当前用户态执行的进程转向内核态执行相关的
异常事件，典型的如缺页异常。
 
        3.外围设备的硬中断  被动的   外围设备完成用户的请求操作后，会像CPU发出中断信号，
此时，CPU就会暂停执行下一条即将要执行的指令，转而去执行中断信号对应的处理程序，
如果先前执行的指令是在用户态下，则自然就发生从用户态到内核态的转换。
 
    参考:https://www.cnblogs.com/bakari/p/5520860.html
        https://blog.csdn.net/qq_34228570/article/details/72995997

进程间通信(IPC)
消息队列（    队列 = 管道 + 锁）
管道(使用消息传递的)
有名管道（FIFO）
信号量
共享内存
套接字（socket）　

管道通常指无名管道
1、它是半双工的（即数据只能在一个方向上流动），具有固定的读端和写端
2、它只能用于具有亲缘关系的进程中通信（也就是父与子进程或者兄弟进程之间）
3、数据不可反复读取了，即读了之后欢喜红区中就没有了
消息队列
1、消息队列是面向记录的，其中的消息具有特定的格式以及特定的优先级
2、消息队列独立于发送与接收进程。进程终止时，消息队列及其内容不会被删除。
3、消息队列可以实现消息随机查询。
 
    队列 = 管道 + 锁

队列,信号量,Event事件,定时器Timer,线程queue,进程池线程池,异步调用+回调机制

同步IO指的是同步传输 ，当发送一个数据请求时，会一直等待，直到有返回结果为止
 
        异步IO指的是异步传输 ，当发送一个数据请求时，会立即去处理别的事情，当有数据
处理完毕后，会自动的返回结果
  
一般同步传输能保证数据正确性 ，而异步能最大化性能。

 
异步IO
    用户进程发起read操作之后，立刻就可以开始去做其它的事。而另一方
面，从kernel的角度，当它受到一个asynchronous read之后，首先它会
立刻返回，所以不会对用户进程产生任何block。然后，kernel会等待数据
准备完成，然后将数据拷贝到用户内存，当这一切都完成之后，kernel会给
用户进程发送一个signal，告诉它read操作完成了。

Value( typecode, arg1, … argN, lock )
 
    在共享内容中常见ctypes对象。typecode要么是包含array模块使用的相同类型代码
（如’i’,’d’等）的字符串，要么是来自ctypes模块的类型对象（如ctypes.c_int、
ctypes.c_double等）。
    所有额外的位置参数arg1, arg2 ….. argN将传递给指定类型的构造函数。lock是只能
使用关键字调用的参数，如果把它置为True（默认值），将创建一个新的锁定来包含对值的访问。
    如果传入一个现有锁定，比如Lock或RLock实例，该锁定将用于进行同步。如果v是Value创建
的共享值的实例，便可使用v.value访问底层的值。例如，读取v.value将获取值，而赋值v.value
将修改值。
 
　　RawValue( typecode, arg1, … ,argN)
同Value对象，但不存在锁定。
 
　　Array( typecode, initializer, lock )
    在共享内存中创建ctypes数组。typecode描述了数组的内容，意义与Value()函数中的相同。
initializer要么是设置数组初始大小的整数，要么是项目序列，其值和大小用于初始化数组。
lock是只能使用关键字调用的参数，意义与Value()函数中相同。
    如果a是Array创建的共享数组的实例，便可使用标准的python索引、切片和迭代操作访问它
的内容，其中每种操作均由锁定进行同步。对于字节字符串，a还具有a.value属性，可以吧整个
数组当做一个字符串进行访问。
 
　　RawArray(typecode, initializer )
    同Array对象，但不存在锁定。当所编写的程序必须一次性操作大量的数组项时，如果同时
使用这种数据类型和用于同步的单独锁定（如果需要的话），性能将得到极大的提升。

import multiprocessing
class FloatChannel(object):
    def __init__(self,maxsize):
        self.buffer=multiprocessing.RawArray(‘d‘,maxsize)
        self.buffer_len=multiprocessing.Value(‘i‘)
        self.empty=multiprocessing.Semaphore(1)
        self.full=multiprocessing.Semaphore(0)
    def send(self,values):
        self.empty.acquire()  #只在缓存为空时继续
        nitems=len(values) 
        self.buffer_len=nitems  #设置缓冲区大小
        self.buffer[:nitems]=values #将复制到缓冲区中
        self.full.release() #发信号通知缓冲区已满
    def recv(self):
        self.full.acquire()     #只在缓冲区已满时继续
        values=self.buffer[:self.buffer_len.value]  #复制值
        self.empty.release()        #发送信号通知缓冲区为空
        return values
    #性能测试 接收多条消息
def consume_test(count,ch):
    for i in xrange(count):
        values=ch.recv()
 
#性能测试 发送多条消息
def produce_test(count,values,ch):
    for i in xrange(count):
        ch.send(values)
if __name__=="__main__":
    ch=FloatChannel(100000)
    p=multiprocessing.Process(target=consume_test,args=(1000,ch))
    p.start()
    values=[float(x) for x in xrange(100000)]
    produce_test(1000,values,ch)
    print "Done"
    p.join()

　　可以通过使用Value或者Array把数据存储在一个共享的内存表中；Manager()返回一个manager类型，控制一个server process，可以允许其它进程通过代理复制一些python objects   支持list,dict,Namespace,Lock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Ｖalue,Array ；

　　Manager类的作用共享资源，manger的的优点是可以在poor进程池中使用，缺点是windows下环境下性能比较差，因为windows平台需要把Manager.list放在if name=‘main‘下，而在实例化子进程时，必须把Manager对象传递给子进程，否则lists无法被共享，而这个过程会消耗巨大资源，因此性能很差。

　　multiprocessing 是一个使用方法类似threading模块的进程模块。允许程序员做并行开发。并且可以在UNIX和Windows下运行。

　　通过创建一个Process 类型并且通过调用call()方法spawn一个进程。

进程间通信：

有两种主要的方式：Queue、Pipe
1- Queue类几乎就是Queue.Queue的复制

2- Pipe 管道

Pipe()返回两个连接类，代表两个方向。如果两个进程在管道的两边同时读或同时写，会有可能造成corruption.

进程间同步
multiprocessing contains equivalents of all the synchronization primitives from threading.
例如，可以加一个锁，以使某一时刻只有一个进程print

进程间共享状态 Sharing state between processes

当然尽最大可能防止使用共享状态，但最终有可能会使用到.
1-共享内存
可以通过使用Value或者Array把数据存储在一个共享的内存表中

‘d‘和‘i‘参数是num和arr用来设置类型，d表示一个双精浮点类型，i表示一个带符号的整型。

更加灵活的共享内存可以使用multiprocessing.sharectypes模块

Server process
　　Manager()返回一个manager类型，控制一个server process，可以允许其它进程通过代理复制一些python objects
支持list,dict,Namespace,Lock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Ｖalue,Array

Server process managers比共享内存方法更加的灵活，一个单独的manager可以被同一网络的不同计算机的多个进程共享。比共享内存更加的缓慢


使用工作池　Using a pool of workers
Pool类代表 a pool of worker processes.
It has methods which allows tasks to be offloaded to the worker processes in a few different ways.

from threading import Thread
import time

def run(name):
    print("%s is running" % name)


if __name__ == "__main__":
    print("主线程")
    time.sleep(10)

    for i in range(10):
        t = Thread(target=run, args=(i,))
        t.start()
        print(time.strftime("%H:%M:%S", time.localtime()))

from concurrent.futures import ThreadPoolExecutor
import time

def run(name):
    print(name)
    time.sleep(3)

if __name__ == "__main__":
    pool = ThreadPoolExecutor(5)
    for i in range(10):
        pool.submit(run, i)

import re
import hashlib
import pickle
 
def func():
    count = 0
    while count<3:
        username = input("username>>>").strip()
        password = input("password>>>").strip()
        if len(password) >=6 and re.search(‘\A([a-z)|[A-Z])‘,password):
            md5_password =  hashlib.md5(password.encode(‘utf-8‘)).hexdigest()
            file_obj = {‘username‘:username,
                        ‘passworf‘:md5_password}
            f = open(‘password.txt‘,‘ab‘)
            pickle.dump(file_obj,f)
            break
        else:
            print("请输入合法的密码")
            count +=1
    else:
        print("机会已经用完")
if __name__ == ‘__main__‘:
    func()　

import socketserver
  
class Handler(socketserver.BaseRequestHandler):
    def handle(self):
        while True:
            try:
                data = self.request.recv(1024)
              #  if not data:
                  #  break
                print(‘client data‘,data.decode("utf-8"))
                self.request.send(data.upper())
            except Exception:
                break
if __name__ == ‘__main__‘:
    server = socketserver.ThreadingTCPServer((‘127.0.0.1‘,8888),Handler)
    server.serve_forever()

import socket
ip_port = (‘127.0.0.1‘,8888)
client = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
client.connect(ip_port)
while True:
    msg = input(">>>>").strip()
    if not msg:
        continue
    client.send(msg.encode(‘utf-8‘))
    data = client.recv(1024)
    print(data)