Добавить многопоточность

@163RUSland · Регистрация: 15.06.2015

Студворк — интернет-сервис помощи студентам

Доброго времени суток! Собрал кое-какой проект (VS 2008, для работы под windows), хочу добавить туда многопоточность, написал кое-что для теста, главный поток отрабатывает нормально:

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
#include <iostream>
#include <fstream>
#include <windows.h>
using namespace std;
 
char ip[14];
char text[256]={};
 
HANDLE logFile;
DWORD WrittenSize;
DWORD dwSize = sizeof(text);
 
void ThreadFunc(fstream &IP)
{
    logFile = CreateFileA("log.txt", GENERIC_WRITE, FILE_SHARE_READ, NULL, OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
    while (!IP.eof()){
        IP >> ip;
        wsprintfA(text, "%s%s", ip, "\r\n");
        SetFilePointer(logFile, 0, NULL, FILE_END);WriteFile(logFile, text, lstrlenA(text), &WrittenSize, NULL);
    }
}
 
int main()
{   
    fstream IP;
    IP.open("ips.txt");
    ThreadFunc(IP);
    IP.close();
    CloseHandle(logFile);
    system("pause");
}[/CODE][/SPOILER]
 
Пытаюсь запустить ThreadFunc() с помощью CreateThread, делаю примерно так:
[CODE]#include <process.h>
#include <iostream>
#include <fstream>
#include <windows.h>
using namespace std;
 
char ip[14];
char text[256]={};
CRITICAL_SECTION cs, cs1;
 
HANDLE logFile;
DWORD WrittenSize;
DWORD dwSize = sizeof(text);
 
DWORD ThreadId;
HANDLE hThread;
 
DWORD WINAPI ThreadFunc(LPVOID lpParametr)
{
    logFile = CreateFileA("log.txt", GENERIC_WRITE, FILE_SHARE_READ, NULL, OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
    InitializeCriticalSection (&cs), InitializeCriticalSection (&cs1);
    while (!IP.eof()){
        EnterCriticalSection( &cs );
        IP >> ip;
        LeaveCriticalSection( &cs );
        wsprintfA(text, "%s%s", ip, "\r\n");
        EnterCriticalSection( &cs1 );
        SetFilePointer(logFile, 0, NULL, FILE_END);WriteFile(logFile, text, lstrlenA(text), &WrittenSize, NULL);
        LeaveCriticalSection( &cs1 );
    }
    DeleteCriticalSection (&cs), (&cs1);
}
 
int main()
{   
    fstream IP;
    IP.open("ips.txt");
    hThread = CreateThread(NULL, 0, ThreadFunc, &IP, 0, &ThreadId);
    WaitForSingleObject(hThread, INFINITE);
    CloseHandle(hThread);
    //ThreadFunc(IP);// вместо ThreadFunc(IP) вставить createthread
    IP.close();
    CloseHandle(logFile);
    system("pause");
}

Получаю:
error C2065: 'IP' : undeclared identifier
error C2228: left of '.eof' must have class/struct/union

Подозреваю, что не правильно передаю массив строк IP и не могу сообразить, как передать его внутрь потоковой функции, уважаемые, подскажите, как это должно выглядеть?

@castaway · 22.06.2015, 13:16

Я немного не так высказался. Давай посмотрим на это с другой стороны...
У тебя каждый поток обрабатывает неопределённое количество адресов. Зачем? Один адрес - один поток.

@163RUSland · 22.06.2015, 13:19 **[ТС]**

castaway, а кто-ж его знает, сколько выпадет адресов на долю каждого потока, тут не угадаешь же, каждый адрес может быть в потоке от долей секунды до нескольких секунд (3-15сек, зависит от того как отработает сокет с адресом), я бы рад выделить каждому адресу по потоку, но боюсь тогда будет куча ложных error connect, сети не тянут обычно более 1-2к одновременных коннектов

@castaway · 22.06.2015, 13:36

Либо я тебя не понимаю, либо ты сам не понимаешь что хочешь сделать..
Моя логика проста: взял адрес - создал поток. Какие ты видишь проблемы и недостатки в этой логике?
Не отвечай размыто и многословно. Ответь кратко и понятно.

@jonson · 22.06.2015, 13:36

Сообщение от 163RUSland

как начинается работа WaitForMultipleObjects, так вылетает

163RUSland, обратите внимание на описание WaitForMultipleObjects, первый параметр nCount определяет количество интересующих Вас объектов ядра, и он не может быть больше MAXIMUM_WAIT_OBJECTS т.е. 64
У Вас как я понял потоков

C++
1
#define THREADS_NUMBER 2000

@163RUSland · 22.06.2015, 13:43 **[ТС]**

castaway, если будет 50к адресов, то создаётся 50к потоков, если все они будут одновременно пытаться подключиться по этим 50 000 ip-адресам, то ошибкой коннекта вернётся более 90-95% из них, поэтому я хочу ограничиться 100-2000 потоков

@castaway · 22.06.2015, 13:52

Мне кажется мы разговариваем на разных языках.
Еще раз: сейчас у тебя один поток обрабатывает неопределённое количество адресов, из-за бесконечного цикла.
Ты понимаешь о чём я говорю?

@163RUSland · 22.06.2015, 14:00 **[ТС]**

castaway, всё верно, каждый поток берёт адреса, пока они не закончатся, если каждому потоку дать строго определенное кол-во адресов это может сказаться на производительности парсинга, опять же, потому что в каких то потоках быстрее закончатся адреса и они будут простаивать без дела.

@castaway · 22.06.2015, 14:17

Сообщение от 163RUSland

если каждому потоку дать строго определенное кол-во адресов

Откуда ты это взял? Один адрес - один поток.

Сообщение от 163RUSland

это может сказаться на производительности парсинга

Мы не обсуждаем парсинг.

@163RUSland · 22.06.2015, 14:27 **[ТС]**

Сообщение от castaway

Откуда ты это взял? Один адрес - один поток.

https://www.cyberforum.ru/post7774257.html ты там писал ещё: Количество потоков соответствует количеству IP-адресов в файле.

@castaway · 22.06.2015, 14:34

Сообщение от castaway

Один адрес - один поток.

Это был мой совет, а не вопрос.

А это:

Сообщение от castaway

Откуда ты это взял?

относится к этому:

Сообщение от 163RUSland

если каждому потоку дать строго определенное кол-во адресов

@163RUSland · 22.06.2015, 14:52 **[ТС]**

castaway, я подумал, что ты предлагаешь дать потокам одинаковое кол-во адресов для работы, исходя из того кода, где каждый поток брал по одному адресу

@castaway · 22.06.2015, 15:05

Я предлагаю дать на каждый поток по одному адресу.

@163RUSland · 22.06.2015, 15:23 **[ТС]**

castaway,не совсем понимаю, можно в виде кода?

@castaway · 22.06.2015, 15:27

Сообщение от 163RUSland

castaway,не совсем понимаю, можно в виде кода?

Ты издеваешься?
У тебя есть код. Там всего то надо убрать бесконечный цикл внутри потока и передавать каждому потоку отдельный IP.

@163RUSland · 22.06.2015, 22:58 **[ТС]**

Сообщение от castaway

Ты издеваешься?

вовсе нет, не имею представления как это будет работать без цикла

@163RUSland · 27.06.2015, 00:46 **[ТС]**

castaway, решил проблему, вставил свой "костыль" вместо WaitForMultipleObjects, тоже своего рода ожидание, после которого можно закрывать потоки, проблема была в цикле, который в потоковой функции, случалось так, что кол-во итераций цикла превышало заданное значение на 1-3 единицы ( в зависимости от размера вектора) в результате переменная int a принимала значения выходящие за вектор, происходило обращение к несуществующему объекту вектора и затем обвал сейчас все работает стабильно:

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
#include "stdafx.h"
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#include <process.h>
#include <iostream>
#include <fstream>
#include <vector>
#include <string>
 
using namespace std;
#define THREADS_NUMBER 5000
 
int a = 0;
 
static CRITICAL_SECTION cs;
 
DWORD WINAPI ThreadFunc(LPVOID lpParametr){
    vector<string> * ips = (vector<string> *)lpParametr;
    char ip[16];
    bool end = 0;
    EnterCriticalSection( &cs );
    if ( a >= ips->size()-3 ) end = 1;
    LeaveCriticalSection( &cs );
    if (end) return 1;
    for (  ; ;  ){
        EnterCriticalSection( &cs );
        if ( a >= ips->size()-3 ) end = 1;
        LeaveCriticalSection( &cs );
        if (end) break;
        EnterCriticalSection( &cs );
        string ip2 = ( *ips )[a];
        strcpy( ip, ip2.c_str() );
        cout << ip << "\n";
        a++;
        //cout << "a:" << a << "\n";
        //cout << "size:" << ips->size() << "\n";
        LeaveCriticalSection( &cs );
        }
    
    return 0;
}
 
int main()
{   
    vector <string> ips;
    fstream fs_ips;
    fs_ips.open( "ips.txt" );
    while ( !fs_ips.eof() ) {
        char ip2[16];
        fs_ips >> ip2;
        ips.push_back( ip2 );
    }
    fs_ips.close();
 
    //if ( THREADS_NUMBER > ips.size() ){THREADS_NUMBER = ips.size();}
    HANDLE * p_hThreads = new HANDLE [THREADS_NUMBER];
    //HANDLE Threads[THREADS_NUMBER];
    InitializeCriticalSection (&cs);
    for( int i = 0; i < THREADS_NUMBER; ++i)
    {
        p_hThreads[i] = CreateThread(NULL, 0, ThreadFunc, &ips, 0, NULL);
        //Threads[i] = CreateThread(NULL, 0, ThreadFunc, &ips, 0, NULL);
    }
    //WaitForMultipleObjects( THREADS_NUMBER, p_hThreads, TRUE, INFINITE );
    bool end2 = 0;
 
    for (  ; ; ){
        Sleep(1000);
        EnterCriticalSection( &cs );
        if ( a >=  ips.size()-3 ) end2 = 1;
        LeaveCriticalSection( &cs );
        if (end2) {
            Sleep(500);
            break;}
    }
 
    for( int i = 0; i < THREADS_NUMBER; i++)
    {
        CloseHandle(p_hThreads[i]);
    }
    DeleteCriticalSection (&cs);
    system("pause");
}

но у меня возникла новая проблема, пока что не понял как, но судя по всему утекает память в потоке, в пределах обработки 15-20к адресов всё в порядке, потоки закрываются и съеденная память освобождается, но дальше начинаются ошибки, программа не вылетает, но всё начинает лететь в error connect, к тому же (палил диспетчер задач), процесс постепенно поглощает всё больше памяти и через секунд 30-40 достигает чуть более 50мб RAM, может нужно выделить под стек по-больше памяти? если можно, то как? Кстати, если не закрывать в этот момент программу, а попытаться произвести подключение к адресу через консоль, то выходит ошибка - "буфер переполнен".

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
#include "stdafx.h"
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#include <iostream>
#include <fstream>
#include <vector>
#include <string>
#include "winsock2.h"
 
using namespace std;
 
char text[256]={};
char text1[256]={};
int a = 0;
 
static CRITICAL_SECTION cs, cs1;
 
HANDLE logFile;
DWORD WrittenSize;
DWORD dwSize = sizeof(text);
 
DWORD WINAPI ThreadFunc(LPVOID lpParametr)
{   
    vector<string> * ips = (vector<string> *)lpParametr;
    char ip[16]={};
    short res;
    SOCKET pop;
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(110);
    
    bool end = 0;
    EnterCriticalSection( &cs );
    if ( a >= ips->size()-3 ) end = 1;
    LeaveCriticalSection( &cs );
    if (end) return 1;
    for (  ; ;  ){
        EnterCriticalSection( &cs );
        if ( a >= ips->size()-3 ) end = 1;
        LeaveCriticalSection( &cs );
        if (end) break;
        EnterCriticalSection( &cs );
        string ip2 = ( *ips )[a];
        strcpy( ip, ip2.c_str() );
        a++;
        LeaveCriticalSection( &cs );
        pop = socket(AF_INET, SOCK_STREAM, 0);
        if(pop == INVALID_SOCKET)
        {
            EnterCriticalSection( &cs1 );
            cout<<"Invalid socket\n";
            wsprintfA(text, "%s%s%s", ip, "_Invalid socket", "\r\n");
            WriteFile(logFile, text, lstrlenA(text), &WrittenSize, NULL);
            memset(text, 0, sizeof(text));
            LeaveCriticalSection( &cs1 );
            closesocket(pop);
            continue;
        }
        addr.sin_addr.s_addr = inet_addr(ip);
 
        BOOL l = TRUE;
        ioctlsocket (pop, FIONBIO, (unsigned long* ) &l);
 
        fd_set read_s;
        timeval time_out;
        FD_ZERO(&read_s);FD_SET(pop, &read_s);time_out.tv_sec = 3;time_out.tv_usec = 0;
 
        connect(pop, (sockaddr *) &addr, sizeof (addr));
        res = select(0, &read_s, NULL, NULL, &time_out);
        if (res <= 0){
            EnterCriticalSection( &cs1 );
            cout << ip << " Error in connect\n";
            wsprintfA(text, "%s%s%s", ip, "[Error in connect or connection timeout]", "\r\n");
            WriteFile(logFile, text, lstrlenA(text), &WrittenSize, NULL);
            memset(text, 0, sizeof(text));
            LeaveCriticalSection( &cs1 );
            closesocket(pop);continue;} 
        if (res > 0){
            EnterCriticalSection( &cs1 );
            cout<< ip << " Connected\n";
            recv(pop, text, (sizeof(text)-1), 0);
            cout<<strlen(text)<<"\n";
            cout<<text<<"\n";
            LeaveCriticalSection( &cs1 );
            if (strlen(text) > 250 || strlen(text) < 3) {
                EnterCriticalSection( &cs1 );
                wsprintfA(text1, "%s%s%s%s%s", ip, "[bad size of answer]", "Response:", text, "\r\n");
                WriteFile(logFile, text1, lstrlenA(text1), &WrittenSize, NULL);
                memset(text, 0, sizeof(text));
                LeaveCriticalSection( &cs1 );
                closesocket(pop);
                continue;}
            EnterCriticalSection( &cs1 );
            wsprintfA(text1, "%s%s%s%s", ip, " ", text, "\n");
            WriteFile(logFile, text1, lstrlenA(text1), &WrittenSize, NULL);
            memset(text, 0, sizeof(text));
            LeaveCriticalSection( &cs1 );
        }
 
        EnterCriticalSection( &cs1 );
        wsprintfA(text, "%s", "quit\r\n");
        send(pop, text, strlen(text), 0); 
        memset(text, 0, sizeof(text));
        LeaveCriticalSection( &cs1 );
        char ip[16]={""};
        }
    closesocket(pop);
    return 0;
}
 
int main()
{   
    int threads_number;
    cout << "amount? ";
    cin >> threads_number;
    vector <string> ips;
    fstream fs_ips;
    fs_ips.open( "ips.txt" );
    while ( !fs_ips.eof() ) {
        char ip2[16];
        fs_ips >> ip2;
        ips.push_back( ip2 );
    }
    fs_ips.close();
    WSADATA wsaData;
    int err;
    err = WSAStartup(MAKEWORD(2,2), &wsaData);
    InitializeCriticalSection (&cs);InitializeCriticalSection (&cs1);
    if(err != 0)
    {
        EnterCriticalSection( &cs1 );
        cout<<"WSAStartup Error: "<<err<<"\n";
        wsprintfA(text, "%s%s%s", "_WSAStartup error: ", err, "\r\n");
        WriteFile(logFile, text, lstrlenA(text), &WrittenSize, NULL);
        LeaveCriticalSection( &cs1 );
        memset(text, 0, sizeof(text));
        WSACleanup();
        return 1;
    }
    HANDLE * p_hThreads = new HANDLE [threads_number];
    //if ( threads_number > ips.size() ) threads_number = ips.size();
    logFile = CreateFileA("log.txt", GENERIC_WRITE, FILE_SHARE_READ, NULL, OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
    for(int i = 0; i < threads_number; i++)
    {
        p_hThreads[i] = CreateThread(NULL, 0, ThreadFunc, &ips, 0, NULL);
    }
    //WaitForMultipleObjects(THREADS_NUMBER, p_hThreads, TRUE, INFINITE);
    bool end2 = 0;
 
    for (  ; ; ){
        Sleep(3000);
        EnterCriticalSection( &cs );
        if ( a >=  ips.size()-3 ) end2 = 1;
        LeaveCriticalSection( &cs );
        if (end2) {
            Sleep(5000);
            break;}
    }
    for(int i = 0; i < threads_number; i++)
    {
        CloseHandle(p_hThreads[i]);
    }
    WSACleanup();
    DeleteCriticalSection (&cs), (&cs1);
    CloseHandle(logFile);
    system("pause");
}

@castaway · 27.06.2015, 07:51

Глядя на код мне просто не хочется с ним разбираться.
Взять к примеру кусок:

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
    int err;
    err = WSAStartup(MAKEWORD(2,2), &wsaData);
    InitializeCriticalSection (&cs);InitializeCriticalSection (&cs1);
    if(err != 0)
    {
        EnterCriticalSection( &cs1 );
        cout<<"WSAStartup Error: "<<err<<"\n";
        wsprintfA(text, "%s%s%s", "_WSAStartup error: ", err, "\r\n");
        WriteFile(logFile, text, lstrlenA(text), &WrittenSize, NULL);
        LeaveCriticalSection( &cs1 );
        memset(text, 0, sizeof(text));
        WSACleanup();
        return 1;
    }

1. Нет удаления критических секций в случае ошибки WSAStartup
2. Какой смысл использовать к/секции без потоков?
3. Если WSAStartup вернула ошибку, значит система сокетов не была инициализирована, следовательно вызов WSACleanup лишний

В самом конце старая история:

C++
1
DeleteCriticalSection (&cs), (&cs1);

Я уже высказывался по этому поводу в этой теме.

Код читать невозможно. Всё в вперемешку, форматирования никакого.

@163RUSland · 27.06.2015, 10:00 **[ТС]**

castaway, теперь вроде читабельно

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
#include "stdafx.h"
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#include <iostream>
#include <fstream>
#include <vector>
#include <string>
#include "winsock2.h"
 
using namespace std;
 
char text[256]={};
char text1[256]={};
int a = 0;
 
CRITICAL_SECTION cs, cs1;
 
HANDLE logFile;
DWORD WrittenSize;
DWORD dwSize = sizeof(text);
 
DWORD WINAPI ThreadFunc(LPVOID lpParametr)
{   
    vector<string> * ips = (vector<string> *)lpParametr;
    char ip[16]={};
    short res = 0;
    SOCKET pop;
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(110);
    
    bool end = 0;
    EnterCriticalSection( &cs );
    if ( a >= ips->size()-3 ) end = 1;
    LeaveCriticalSection( &cs );
    if (end) return 1;
    for (  ; ;  ){
        EnterCriticalSection( &cs );
        if ( a >= ips->size()-3 ) end = 1;
        LeaveCriticalSection( &cs );
        if (end) break;
        EnterCriticalSection( &cs );
        string ip2 = ( *ips )[a];
        strcpy( ip, ip2.c_str() );
        a++;
        LeaveCriticalSection( &cs );
        pop = socket(AF_INET, SOCK_STREAM, 0);
        if(pop == INVALID_SOCKET)
        {
            EnterCriticalSection( &cs1 );
            cout << "Invalid socket\n";
            wsprintfA(text, "%s%s%s", ip, "_Invalid socket", "\r\n");
            WriteFile(logFile, text, lstrlenA(text), &WrittenSize, NULL);
            memset(text, 0, sizeof(text));
            LeaveCriticalSection( &cs1 );
            closesocket(pop);
            continue;
        }
        addr.sin_addr.s_addr = inet_addr(ip);
 
        BOOL l = TRUE;
        ioctlsocket (pop, FIONBIO, (unsigned long* ) &l);
 
        fd_set read_s;
        timeval time_out;
        FD_ZERO(&read_s);
        FD_SET(pop, &read_s);
        time_out.tv_sec = 3;
        time_out.tv_usec = 0;
 
        connect(pop, (sockaddr *) &addr, sizeof (addr));
        res = select(0, &read_s, NULL, NULL, &time_out);
        if (res <= 0){
            EnterCriticalSection( &cs1 );
            cout << ip << " Error in connect\n";
            wsprintfA(text, "%s%s%s", ip, "[Error in connect or connection timeout]", "\r\n");
            WriteFile(logFile, text, lstrlenA(text), &WrittenSize, NULL);
            memset(text, 0, sizeof(text));
            LeaveCriticalSection( &cs1 );
            closesocket(pop);
            continue;
        } 
        if (res > 0){
            EnterCriticalSection( &cs1 );
            cout<< ip << " ";
            recv(pop, text, (sizeof(text)-1), 0);
            cout << "response size:" << strlen(text) << " " << text;
            LeaveCriticalSection( &cs1 );
            if (strlen(text) > 250 || strlen(text) < 3) {
                EnterCriticalSection( &cs1 );
                wsprintfA(text1, "%s%s%s%s%s", ip, "[bad size of answer]", "Response:", text, "\r\n");
                WriteFile(logFile, text1, lstrlenA(text1), &WrittenSize, NULL);
                memset(text, 0, sizeof(text));
                LeaveCriticalSection( &cs1 );
                closesocket(pop);
                continue;
            }
            EnterCriticalSection( &cs1 );
            wsprintfA(text1, "%s%s%s%s", ip, " ", text, "\n");
            WriteFile(logFile, text1, lstrlenA(text1), &WrittenSize, NULL);
            memset(text, 0, sizeof(text));
            LeaveCriticalSection( &cs1 );
        }
 
        EnterCriticalSection( &cs1 );
        wsprintfA(text, "%s", "quit\r\n");
        send(pop, text, strlen(text), 0); 
        memset(text, 0, sizeof(text));
        LeaveCriticalSection( &cs1 );
        char ip[16]={""};
        }
    closesocket(pop);
    return 0;
}
 
int main()
{   
    int threads_number;
    cout << "amount? ";
    cin >> threads_number;
    vector <string> ips;
    fstream fs_ips;
    fs_ips.open( "ips.txt" );
    while ( !fs_ips.eof() ) {
        char ip2[16];
        fs_ips >> ip2;
        ips.push_back( ip2 );
    }
    fs_ips.close();
    WSADATA wsaData;
    short err;
    err = WSAStartup(MAKEWORD(2,2), &wsaData);
    if(err != 0)
    {
        cout<<"WSAStartup Error: "<<err<<"\n";
        wsprintfA(text, "%s%s%s", "_WSAStartup error: ", err, "\r\n");
        WriteFile(logFile, text, lstrlenA(text), &WrittenSize, NULL);
        memset(text, 0, sizeof(text));
        return 1;
    }
    HANDLE * p_hThreads = new HANDLE [threads_number];
    if ( threads_number > ips.size() ) threads_number = ips.size();
    logFile = CreateFileA("log.txt", GENERIC_WRITE, FILE_SHARE_READ, NULL, OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
    InitializeCriticalSection (&cs);
    InitializeCriticalSection (&cs1);
    for(int i = 0; i < threads_number; i++)
    {
        p_hThreads[i] = CreateThread(NULL, 0, ThreadFunc, &ips, 0, NULL);
    }
    bool end2 = 0;
 
    for (  ; ; ){
        Sleep(3000);
        EnterCriticalSection( &cs );
        if ( a >=  ips.size()-3 ) end2 = 1;
        LeaveCriticalSection( &cs );
        if (end2) {
            Sleep(5000);
            break;}
    }
    for(int i = 0; i < threads_number; i++)
    {
        CloseHandle(p_hThreads[i]);
    }
    WSACleanup();
    DeleteCriticalSection (&cs);
    DeleteCriticalSection (&cs1);
    CloseHandle(logFile);
    system("pause");
}

остальное планировал привести в порядок в конце самом, не подумал, что читать трудно из-за этого, извиняюсь за бардак.

@castaway · 27.06.2015, 10:36

Опиши в двух словах алгоритм работы программы.

@163RUSland · 27.06.2015, 10:50 **[ТС]**

castaway, из ips.txt загружаются в вектор адреса (50к и более), далее потоки берут с вектора адреса и подключаются к ним через 110 (pop) порт, на консоль выводятся построчно ошибки соединения, либо текущий ip-адрес, размер ответа и ответ (если он есть), всё это попутно пишется в файл, кроме размера ответа.

Новые блоги и статьи Все статьи Все блоги /
Море в объективе kumehtar 06.07.2026 Хотел написать много проникновенных слов, но передумал. Оно само - говорит. Кто слышит тот слышит.	Оказывается, Unreal Engine позволяет качество на порядки выше, чем было в Lineedge Etyuhibosecyu 05.07.2026 Жаль, конечно, что я не узнал об этом, пока Lineedge существовала, а то бы Noname2331 написал, что волки превращаются в пиксельную кашу, а я бы его попросил скачать какую-нибудь бриллиантовую или Pro. . .	Doom для терминала без стрельбы и монстров. 3D Raycasting на ascii. dcc0 05.07.2026 Попросил нейронную сеть deepai. org написать рейкастинг 3D с библиотекой ncurses для Linux. Чтобы можно было ходить на стрелочки. Чтобы стены были отрисованы символами. Справилась. Первый вариант. . .	Установка статуса документа по условию Maks 05.07.2026 Алгоритм из решения ниже реализован на нетиповом документе "НарядПутевка" разработанного в КА2. Задача: в табличной части "Материалы" документа при записи автоматически устанавливать статус. . .
Сезонность и суточность закисления почв anaschu 04.07.2026 200 часов это все равно моловато. Есть ситуации, но нестандартные, когда смена происходит за 5 лет. Но обычно это 50 лет и более. Наверное, закисление почвы происходит сезонно в средней. . .	В чем ценность человеческого опыта в глобальном смысле? kumehtar 03.07.2026 Возможно, ценность человека не в том, что он однажды достигает мудрости, а в том, что он становится носителем карты пути. Он знает не только истину, но и последовательность внутренних изменений,. . .	интеграция AnyLogic с самописным REST API и переход на Odoo anaschu 03.07.2026 Успешная интеграция AnyLogic с самописным REST API и переход на промышленную Odoo WMS Сегодня проделал огромный путь от простой симуляции физических процессов до построения полноценной. . .	Поиск всех путей на ориентированном графе. Linux dcc0 02.07.2026 Переработка старого кода из моей статьи. Через несколько переработок от PHP кода к C89 (надеюсь, 89). Но довольно запутанно получилось. Код для Linux. Но если убрать time и то, что с ним. . .

@castaway 4986 / 3093 / 456 Регистрация: 10.11.2010 Сообщений: 11,170 Записей в блоге: 10
	22.06.2015, 13:16
	Я немного не так высказался. Давай посмотрим на это с другой стороны... У тебя каждый поток обрабатывает неопределённое количество адресов. Зачем? Один адрес - один поток. 0

@163RUSland 0 / 0 / 0 Регистрация: 15.06.2015 Сообщений: 42
	22.06.2015, 13:19 [ТС]
	castaway, а кто-ж его знает, сколько выпадет адресов на долю каждого потока, тут не угадаешь же, каждый адрес может быть в потоке от долей секунды до нескольких секунд (3-15сек, зависит от того как отработает сокет с адресом), я бы рад выделить каждому адресу по потоку, но боюсь тогда будет куча ложных error connect, сети не тянут обычно более 1-2к одновременных коннектов 0

@castaway 4986 / 3093 / 456 Регистрация: 10.11.2010 Сообщений: 11,170 Записей в блоге: 10
	22.06.2015, 13:36
	Либо я тебя не понимаю, либо ты сам не понимаешь что хочешь сделать.. Моя логика проста: взял адрес - создал поток. Какие ты видишь проблемы и недостатки в этой логике? Не отвечай размыто и многословно. Ответь кратко и понятно. 0

@163RUSland 0 / 0 / 0 Регистрация: 15.06.2015 Сообщений: 42
	22.06.2015, 13:43 [ТС]
	castaway, если будет 50к адресов, то создаётся 50к потоков, если все они будут одновременно пытаться подключиться по этим 50 000 ip-адресам, то ошибкой коннекта вернётся более 90-95% из них, поэтому я хочу ограничиться 100-2000 потоков 0

@castaway 4986 / 3093 / 456 Регистрация: 10.11.2010 Сообщений: 11,170 Записей в блоге: 10
	22.06.2015, 13:52
	Мне кажется мы разговариваем на разных языках. Еще раз: сейчас у тебя один поток обрабатывает неопределённое количество адресов, из-за бесконечного цикла. Ты понимаешь о чём я говорю? 0

@163RUSland 0 / 0 / 0 Регистрация: 15.06.2015 Сообщений: 42
	22.06.2015, 14:00 [ТС]
	castaway, всё верно, каждый поток берёт адреса, пока они не закончатся, если каждому потоку дать строго определенное кол-во адресов это может сказаться на производительности парсинга, опять же, потому что в каких то потоках быстрее закончатся адреса и они будут простаивать без дела. 0

@163RUSland 0 / 0 / 0 Регистрация: 15.06.2015 Сообщений: 42
	22.06.2015, 14:52 [ТС]
	castaway, я подумал, что ты предлагаешь дать потокам одинаковое кол-во адресов для работы, исходя из того кода, где каждый поток брал по одному адресу 0

@castaway 4986 / 3093 / 456 Регистрация: 10.11.2010 Сообщений: 11,170 Записей в блоге: 10
	22.06.2015, 15:05
	Я предлагаю дать на каждый поток по одному адресу. 0

@163RUSland 0 / 0 / 0 Регистрация: 15.06.2015 Сообщений: 42
	22.06.2015, 15:23 [ТС]
	castaway,не совсем понимаю, можно в виде кода? 0

@castaway 4986 / 3093 / 456 Регистрация: 10.11.2010 Сообщений: 11,170 Записей в блоге: 10
	27.06.2015, 10:36
	Опиши в двух словах алгоритм работы программы. 0

@163RUSland 0 / 0 / 0 Регистрация: 15.06.2015 Сообщений: 42
	27.06.2015, 10:50 [ТС]
	castaway, из ips.txt загружаются в вектор адреса (50к и более), далее потоки берут с вектора адреса и подключаются к ним через 110 (pop) порт, на консоль выводятся построчно ошибки соединения, либо текущий ip-адрес, размер ответа и ответ (если он есть), всё это попутно пишется в файл, кроме размера ответа. 0