async server u zig-u (sve platforme)
Elem opet sam se malo igrao za zigom, pa napravih mali server, koji donekle dobro radi.
Radi ovako: primi do 100 konekcija, pa onda ceka da se sve zavrse...
/data/assets/logo/forum_192x192.png
Forum Krstarice
import Foundation signal(SIGPIPE,SIG_IGN) let s = Server(); s.start() var factor = 4 var ende:Bool? = nil var semacnt:Int? = nil var occupied:Int? = nil var timeout:timespec? = nil var timeoutr:Int? = nil var timeoutw:Int? = nil var serviced:Int? = nil var accarrFD:[Int32]? var cond:NSCondition? var lock:NSLock? var standardError:FileHandle? = nil class Server { init() { let task = Process() task.executableURL = URL(fileURLWithPath: "/usr/sbin/sysctl") task.arguments = ["hw.ncpu"] let out = Pipe() task.standardOutput = out let res: ()? = try? task.run() if let _ = res { let outputData = out.fileHandleForReading.readDataToEndOfFile() let s = String(decoding: outputData, as: UTF8.self) print("got from sysctl ",s) let sep = CharacterSet(charactersIn: " \n") let components = s.components(separatedBy: sep) factor = Int(components[1]) ?? 4 } if (CommandLine.argc > 1) { tmpSvcPort = Int(CommandLine.arguments[1]) ?? 0 if (tmpSvcPort > 0){ servicePort = String(tmpSvcPort) } } semacnt = 0 occupied = 0 ende = false timeoutr = 0 timeoutw = 0 serviced = 0 timeout = timespec(tv_sec:5,tv_nsec:0) standardError = FileHandle.standardError cond = NSCondition() lock = NSLock() accarrFD = Array(repeating:Int32(0),count:0) } deinit { } var tmpSvcPort: Int = 0 var servicePort = "1234" class Worker:Thread{ override init(){ super.init() semacnt! += 1 } deinit { print("deinit") semacnt! -= 1 } override func main(){ let MTU = 65536 let buffer = UnsafeMutablePointer<CChar>.allocate(capacity:MTU) defer { buffer.deallocate() } let sockKqueue = kqueue() if sockKqueue == -1 { print("Error creating kqueue",to:&standardError!) } var servicedone = 0 defer {close(sockKqueue)} while(semacnt! <= factor && servicedone<100000){ cond!.lock() while (accarrFD!.isEmpty){ cond!.wait() } var oneFD:[Int32:kevent]=[:] var count = 0 while (!accarrFD!.isEmpty && count<factor){ /// collect rest of fds let fd = accarrFD![accarrFD!.count-1] oneFD[fd]=kevent( ident: UInt(fd), filter: Int16(EVFILT_READ), flags: UInt16(EV_ADD | EV_ENABLE), fflags: 0, data: 0, udata: nil) accarrFD!.remove(at:accarrFD!.count-1) count+=1 } cond!.unlock() lock!.lock() occupied! += 1 serviced!+=oneFD.count servicedone+=oneFD.count lock!.unlock() while !oneFD.isEmpty{ let toproc = oneFD var arrFD = Array(repeating:kevent(),count:0) for (fd,value) in toproc{ let flags = fcntl(fd,F_GETFL) let _ = fcntl(fd,F_SETFL,flags | O_NONBLOCK) arrFD.append(value) } kevent(sockKqueue, &arrFD, Int32(arrFD.count), nil, 0, nil) var event = Array(repeating:kevent(),count:arrFD.count) let status = kevent(sockKqueue, nil, 0, &event, Int32(event.count), &timeout!) if status == 0 { print("Timeout") timeoutr! += 1 for i in oneFD { close(i.key) } oneFD.removeAll() continue } else if status > 0 { for i in 0..<status { let i:Int = Int(i) let fd:Int32 = Int32(event[i].ident) defer {close(fd);oneFD.removeValue(forKey:fd)} if (event[i].flags & UInt16(EV_EOF)) == EV_EOF { print("The socket (\(event[i].ident)) has been closed.") continue } print("File descriptor: \(event[i].ident) - has \(event[i].data) characters for reading") let readResult = read(fd, buffer, event[i].data>MTU ? MTU : event[i].data) if (readResult == 0) { continue; // end of file } else if (readResult == -1) { print("Error reading form client(\(fd)) - \(errno)",to:&standardError!) continue; // error } else { buffer[event[i].data]=0 let strResult = String(cString: buffer) let cset = CharacterSet(charactersIn: " \r\n") let numbers = strResult.components(separatedBy: cset) var sum = 0 if numbers.count > 1 { if numbers[0].lowercased() == "quit" { ende! = true continue } else { let num1 = Int(numbers[0]) ?? 0 let num2 = Int(numbers[1]) ?? 0 sum = num1 + num2 } } print("Received form client(\(fd)): \(strResult)") let out = String(format:"suma %d\n",sum) let _ = out.withCString { var sockKevent = kevent( ident: UInt(fd), filter: Int16(EVFILT_WRITE), flags: UInt16(EV_ADD | EV_ENABLE), fflags: 0, data: 0, udata: nil) kevent(sockKqueue, &sockKevent, 1, nil, 0, nil) var event = kevent() let status = kevent(sockKqueue, nil, 0, &event, 1, &timeout!) if status == 0 { print("Timeout") timeoutw!+=1 } else if status > 0 { if (event.flags & UInt16(EV_EOF)) == EV_EOF { print("The socket (\(fd)) has been closed.") } print("File descriptor: \(fd) - can write") write(fd, $0, out.count) } else { print("Error reading kevent") } } } } } else { print("Error reading kevent",to:&standardError!) continue } } print("release occupied",occupied!,semacnt!,timeoutr!,timeoutw!,serviced!) lock!.lock() occupied! -= 1 lock!.unlock() } } } func start() { print("Server starting...") let socketFD = socket(AF_INET6, //Domain [AF_INET,AF_INET6, AF_UNIX] SOCK_STREAM, //Type [SOCK_STREAM, SOCK_DGRAM, // SOCK_SEQPACKET, SOCK_RAW] IPPROTO_TCP //Protocol [IPPROTO_TCP, IPPROTO_SCTP, // IPPROTO_UDP, IPPROTO_DCCP] ) //Return a FileDescriptor -1 = error if socketFD == -1 { print("Error creating BSD Socket",to:&standardError!) return } var hints = addrinfo( ai_flags: AI_PASSIVE, // Assign the address of the local host to the // socket structures ai_family: AF_UNSPEC, // Either IPv4 or IPv6 ai_socktype: SOCK_STREAM,// TCP ai_protocol: 0, ai_addrlen: 0, ai_canonname: nil, ai_addr: nil, ai_next: nil) var servinfo: UnsafeMutablePointer<addrinfo>? = nil let addrInfoResult = getaddrinfo( nil, // Any interface servicePort, // The port on which will be listenend &hints, // Protocol configuration as per above &servinfo) if addrInfoResult != 0 { print("Error getting address info: \(errno)",to:&standardError!) return } setsockopt(socketFD,SOL_SOCKET,SO_REUSEADDR,servinfo!.pointee.ai_addr, socklen_t(servinfo!.pointee.ai_addrlen)) let bindResult = bind(socketFD, servinfo!.pointee.ai_addr, socklen_t(servinfo!.pointee.ai_addrlen)) if bindResult == -1 { print("Error binding socket to Address: \(errno)",to:&standardError!) return } let listenResult = listen(socketFD, //Socket File descriptor 128 // adequate // The backlog argument defines the // maximum length the queue of pending // connections may grow to ) let flags = fcntl(socketFD,F_GETFL) let _ = fcntl(socketFD,F_SETFL,flags | O_NONBLOCK) let sockKqueue = kqueue() if sockKqueue == -1 { print("Error creating kqueue",to:&standardError!) } var sockKevent = kevent( ident: UInt(socketFD), filter: Int16(EVFILT_READ), flags: UInt16(EV_ADD | EV_ENABLE), fflags: 0, data: 0, udata: nil) kevent(sockKqueue, &sockKevent, 1, nil, 0, nil) var event = kevent() if listenResult == -1 { print("Error setting our socket to listen",to:&standardError!) return } while (!ende!) { var addr = sockaddr() var addr_len :socklen_t = 0 while (!ende!) { print("About to accept") let status = kevent(sockKqueue, nil, 0, &event, 1, nil) if status == 0 { print("Timeout") continue } else if status > 0 { if (event.flags & UInt16(EV_EOF)) == EV_EOF { print("The socket (\(socketFD)) has been closed.shutdown.",to:&standardError!) exit(0) } print("File descriptor: \(socketFD) - has \(event.data) characters for reading") } else { print("Error reading kevent",to:&standardError!) continue } for _ in 0..<event.data { let accFD = accept(socketFD, &addr, &addr_len) if (accFD > 0){ print("Accepted new client with file descriptor: \(accFD)") cond!.lock() accarrFD!.append(accFD) if (semacnt! < factor || semacnt! <= occupied!) { let handle = Worker(); handle.start() } cond!.unlock() } else { print("Error accepting connection",to:&standardError!);break } } cond!.broadcast() } } exit(EXIT_SUCCESS) } } extension FileHandle: TextOutputStream { public func write(_ string: String) { let data = Data(string.utf8) self.write(data) } }