marcelb
/
socket
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Work on queue..

Browse Source
queue
marcelb 12 months ago
parent 9217ce91c4
commit b0cfca949e
8 changed files with 276 additions and 38 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 3
      .vscode/settings.json
  2. 26
      lib/tcp_socket.hpp
  3. 168
      src/tcp_socket.cpp
  4. 95
      test/client.cpp
  5. BIN
      test/client.o
  6. 2
      test/compile-server.sh
  7. 20
      test/server.cpp
  8. BIN
      test/server.o

3
.vscode/settings.json vendored
Unescape View File

@ -45,6 +45,7 @@
"streambuf": "cpp",
"cinttypes": "cpp",
"typeinfo": "cpp",
"thread": "cpp"
"thread": "cpp",
"chrono": "cpp"
}
}

26
lib/tcp_socket.hpp
Unescape View File

@ -4,6 +4,7 @@
#include <iostream>
#include <string>
#include <vector>
#include <deque>
#include <thread>
#include <mutex>
#include <string.h>
@ -102,4 +103,29 @@ class client {
string pull (size_t byte_limit = 1024);
};
class queue { // FIFO
public:
mutex io;
uint limit;
string address;
ushort port;
uint timeout;
server *srv = NULL;
SSL_CTX* securefds = NULL;
deque<client*> clients;
thread controller;
queue (const uint _limit, const string _address, const ushort _port, const uint _timeout = 100, SSL_CTX* _securefds = NULL);
queue (const uint _limit, server *_srv, const uint _timeout = 100, SSL_CTX* _securefds = NULL);
~queue();
void enqueue();
client dequeue();
};
#endif

168
src/tcp_socket.cpp
Unescape View File

@ -258,9 +258,6 @@ client::client(const server *_srv, const uint timeout, SSL_CTX* securefds) {
}
#endif
if (securefds) {
ssl = SSL_new(securefds);
if (!ssl) {
@ -356,3 +353,168 @@ string client::pull (size_t byte_limit) {
queue::queue (const uint _limit, const string _address, const ushort _port, const uint _timeout, SSL_CTX* _securefds) {
if (_limit > 1) {
limit = _limit;
}
else {
throw string("[ERROR] Limit out of range ");
}
address = _address;
port = _port;
timeout = _timeout;
securefds = _securefds;
// for (uint i=0; i<limit; i++) {
// if (address.empty() && srv != NULL) {
// clients.push_back(new client(srv, timeout, securefds));
// }
// else {
// clients.push_back(new client(address, port, timeout, securefds));
// }
// }
// controller = thread([&](){
// cout << "Upao u tred "<< endl;
// while(true) {
// // io.lock();
// // uint missing = limit-clients.size();
// // cout << "nedostaje " << missing << endl;
// // if (missing > 0.6*limit) {
// // enqueue();
// if (clients.size() <= 1) {
// if (address.empty() && srv != NULL) {
// clients.push_back(new client(srv, timeout, securefds));
// }
// else {
// clients.push_back(new client(address, port, timeout, securefds));
// }
// }
// // }
// // usleep(missing*100000);
// // io.unlock();
// // usleep(1000);
// sleep(1);
// }
// });
}
queue::queue (const uint _limit, server *_srv, const uint _timeout, SSL_CTX* _securefds) {
if (_limit > 1) {
limit = _limit;
}
else {
throw string("[ERROR] Limit out of range ");
}
srv = _srv;
timeout = _timeout;
securefds = _securefds;
controller = thread([this](){
cout << "Upao u tred "<< endl;
while(true) {
io.lock();
// uint missing = limit-clients.size();
// cout << "nedostaje " << missing << endl;
// if (missing > 0.6*limit) {
// enqueue();
if (clients.size()< limit) {
if (address.empty() && srv != NULL) {
clients.push_back(new client(srv, timeout, securefds));
}
else {
clients.push_back(new client(address, port, timeout, securefds));
}
}
// }
// usleep(missing*100000);
io.unlock();
// usleep(1000);
}
});
}
void queue::enqueue () {
vector<thread> worker;
lock_guard<mutex> master(io);
for (uint i=0; i<limit-clients.size(); i++) {
cout << "idemo dodati klijente " << endl;
worker.push_back(thread([&](){
if (address.empty() && srv != NULL) {
clients.push_back(new client(srv, timeout, securefds));
}
else {
cout << "dodajem " << i << endl;
clients.push_back(new client(address, port, timeout, securefds));
}
}));
}
for (uint i=0; i<worker.size(); i++) {
cout << "čekam da se dodaju " << endl;
worker[i].join();
}
cout << "dodali su se " << endl;
worker.clear();
}
client queue::dequeue() {
cout << "uzimam jednog iz stacka " << endl;
lock_guard<mutex> master(io);
client *cli;
while (true) {
if (clients.size() >= 1) {
cli = clients.front();
clients.pop_front();
cout << "uzimam jednog iz stacka uzeo sam " << endl;
return *cli;
}
else if (clients.size() == 1) {
controller = thread([&](){
cout << "Upao u tred "<< endl;
while(true) {
// io.lock();
// uint missing = limit-clients.size();
// cout << "nedostaje " << missing << endl;
// if (missing > 0.6*limit) {
// enqueue();
if (clients.size() <= 1) {
if (address.empty() && srv != NULL) {
clients.push_back(new client(srv, timeout, securefds));
}
else {
clients.push_back(new client(address, port, timeout, securefds));
}
}
// }
// usleep(missing*100000);
// io.unlock();
// usleep(1000);
sleep(1);
}
});
}
usleep(1000);
}
}
queue::~queue () {
for (uint i=0; i<clients.size(); i++) {
clients[i]->~client();
}
clients.clear();
}

95
test/client.cpp
Unescape View File

@ -1,44 +1,93 @@
#include <iostream>
#include <string>
#include <chrono>
#include "../lib/tcp_socket.hpp"
using namespace std;
using namespace chrono;
int main() {
try {
cout << "inicijalizacija reda" << endl;
// queue sockets(2, "127.0.0.1", 5000, 500);
uint n = 10000;
vector<thread> thr;
for (uint i=0; i<n; i++) {
thr.push_back(thread([](uint a){
client myserver("127.0.0.1", 5000, 500);
string sends = "Hello world " + to_string(a);
myserver.push(sends);
cout << myserver.pull() << endl;
}, i));
}
// thread t1([&]() {
// cout << "korištenje prvog klijenta" << endl;
// int a = 0;
// while(true) {
// try {
// auto start = high_resolution_clock::now();
// client myserver = sockets.dequeue();
// auto stop = high_resolution_clock::now();
// auto duration = duration_cast<microseconds>(stop - start);
// cout << "Time taken by function: " << duration.count() << " microseconds" << endl;
// string sends = "Socket I " + to_string(a++);
// myserver.push(sends);
// cout << myserver.pull() << endl;
for (uint i=0; i<n; i++) {
thr[i].join();
}
// } catch (const string err) {
// cout << err << endl;
// }
// // usleep(100000);
// }
// });
// // usleep(20000);
// secure crypto;
// cout << "init cert " << endl;
// thread t2([&]() {
// cout << "korištenje drugog klijenta" << endl;
// int a = 0;
// while(true) {
// try {
// auto start = high_resolution_clock::now();
// client myserver = sockets.dequeue();
// auto stop = high_resolution_clock::now();
// auto duration = duration_cast<microseconds>(stop - start);
// cout << "Time taken by function: " << duration.count() << " microseconds" << endl;
// string sends = "Socket II " + to_string(a++);
// myserver.push(sends);
// cout << myserver.pull() << endl;
// client myserver("127.0.0.1", 5000, 5000, crypto.fds);
client myserver("localhost", 8000, 5000, crypto.fds);
// client myserver("localhost", 5000);
// cout << "init client " << endl;
// } catch (const string err) {
// cout << err << endl;
// }
// // usleep(100000);
// }
// });
// t1.join();
// t2.join();
// string sends = "Hello world!";
// cout << myserver.push(sends) << " " << sends.length() << endl;
// cout << "wait client " << endl;
// int a = 0;
// auto start = high_resolution_clock::now();
// client myserver ("127.0.0.1", 5000, 500);
// // auto stop = high_resolution_clock::now();
// // auto duration = duration_cast<microseconds>(stop - start);
// // cout << "Time taken by function: " << duration.count() << " microseconds" << endl;
// string sends = "Socket II " + to_string(a++);
// myserver.push(sends);
// cout << myserver.pull() << endl;
// auto stop = high_resolution_clock::now();
// auto duration = duration_cast<microseconds>(stop - start);
// cout << "Time taken by function: " << duration.count() << " microseconds" << endl;
vector<client*> clis;
for (uint i =0 ; i<5; i++) {
clis.push_back(new client("127.0.0.1", 5000, 500));
}
// cout << myserver.pull();
cout << "Dodalo je đubre "<< endl;
sleep(5);
string sends = "Socket I ";
clis[1]->push(sends);
cout << clis[1]->pull() << endl;
sends = "Socket III ";
clis[3]->push(sends);
cout << clis[3]->pull() << endl;
}
catch (const string err) {

BIN
test/client.o
View File

Binary file not shown.

2
test/compile-server.sh
Unescape View File

@ -1 +1 @@
g++ server.cpp ../src/* -o server.o -lssl -lcrypto
g++ server.cpp ../src/* -o server.o -lssl -lcrypto -lpthread

20
test/server.cpp
Unescape View File

@ -10,10 +10,10 @@ int main() {
// secure crypto ("../example/cert.pem", "../example/privkey.pem");
// cout << "init server " << endl;
// server myserver(5000, 100, crypto.fds);
cout << "init cert " << endl;
secure crypto ("../example/cert.pem", "../example/privkey.pem");
cout << "init server " << endl;
server myserver(8000, 100, crypto.fds);
// cout << "init cert " << endl;
// secure crypto ("../example/cert.pem", "../example/privkey.pem");
// cout << "init server " << endl;
// server myserver(8000, 100, crypto.fds);
// cout << "init client " << endl;
@ -29,12 +29,12 @@ int main() {
// // usleep(600*1000);
// sleep(5);
// myserver.cli->push(fromclient);
string fromclient = myserver.cli->pull();
// string fromclient = myserver.cli->pull();
// string fromclient = myclient.pull();
cout << "tell client " << fromclient << endl;
// cout << "tell client " << fromclient << endl;
// usleep(600*1000);
//sleep(5);
myserver.cli->push(fromclient);
// myserver.cli->push(fromclient);
// myclient.push(fromclient);
// myclient.~comming();
@ -54,12 +54,12 @@ int main() {
myserver.async(8, [](client &cli, mutex &io) {
cout << "Klijent " << cli.ipv4 << endl;
string fromclient = cli.pull();
io.lock();
// io.lock();
cout << "S klijenta " << fromclient << endl;
io.unlock();
// io.unlock();
// fromclient += teststr;
cli.push(fromclient);
}, 200);
}, 0);
// string teststr = " Idemooo";

BIN
test/server.o
View File

Binary file not shown.
Write Preview
Loading…
Cancel
Save