Здравствуйте.

У меня задача: найти на видео-изображении все координаты темных мест (например прорыв бумаги). Написал алгоритм, где часть написана на ассемблере (MMX), а часть на C++ (вызывается функция из ассемблерного кода). Просто проход по всем точкам изображения, предположим, занимает 100 мс (очень большое изображение). У меня двух процессорная машина; после распараллеливания этого алгоритма на два потока — время составляет около 60 мс. Дальше — сохраняю эти координаты в векторе в однопоточном режиме работы — стало 600 мс (добавилась логика и т.д.).
Кодга я запустил алгоритм в двух потоках, причем создал по вектору для каждого потока, скорость упала до 14 секунд, хотя я использовал метод vector::reserve() для каждого вектора, и количество добавляемых элементов никак не превышает предел, указанный в методе reserve().
Может ли быть ситуация, что STL синхронизирует обращение к векторам (т.е. v1->push_back() и v2->push_back() не могут выполниться паралелльно, хотя и находятся в разных потоках).

Здравствуйте, dmitry_filinov, Вы писали:

_>...

В стандарте С++, как я знаю, ничего по поводу потоковой синхронизации контейнеров stl не говорится
По идее, операция с одним контейнером(vector) никак не должна влиять на операции со вторым контейнером(vector)
Единственное, что может существенно тормозить программу — это многократные выделения памяти. Но вы сказали,
что предварительно выделяете всю необходимую память для ваших контейнеров. Это точно так!?
Может быть у вас есть ещё какая-то ваша синхронизация между выполняющимися потоками!?

Здравствуйте, dmitry_filinov, Вы писали:

_>Здравствуйте...

Могу ошибаться, но возможно дело в кеш-линиях. Ну и постоянная загрузка-выгрузка контекста потока с такими объемами ничего хорошего не сулит. У Вас вектора как локальные переменные заданы ?

Я выделяю память на старте с помощью метода reserve(300000), фактически получается около 200000 элементов. Мне, просто, очень лень писать свою реализацию вектора, но, видимо, придется, т.к. не могу больше ни на что сетовать ...
Высылаю примеры исходника, может, что увидите неверного

вот примеры исходника ....

перед запуском потоков инициализирую вектора ...
typedef struct {
int start_x;
int end_x;
int y;
} TRowBrak;
typedef std::vector<TRowBrak> TRowBrakVec;

.///////////////////////////////////////////////////////////////

for (int i=0;i<cpu_core_count;i++) {
lp_threads[i].lst_brak=new TRowBrakVec();
lp_threads[i].lst_brak->reserve(300000);

.....
}

.///////////////////////////////////////////////////////////////
а вот потоковая функция обработки ...
void add_brak_to_list(TRowBrakVec *lst_brak, BYTE *pix_vals_arr, BYTE vals_count,
int x, int y, BYTE *is_start_brak, int y_shift)
{
//Sleep(1);
//return;

#define PROCESS_BRAK_POINT(point_x) \
if (*is_start_brak==1) { \
int lst_brak_size=lst_brak->size(); \
if (lst_brak_size!=0) { \
TRowBrak &prior_brak=lst_brak->end()[-1]; \
if ((prior_brak.y==y)&&((point_x-prior_brak.end_x)<=3)) \
prior_brak.end_x=point_x; \
else { \
TRowBrak brak; \
brak.end_x=0; \
brak.start_x=point_x; \
brak.y=y; \
lst_brak->push_back(brak); \
} \
} else { \
TRowBrak brak; \
brak.end_x=0; \
brak.start_x=point_x; \
brak.y=y;\
lst_brak->push_back(brak); \
} \
} else { \
TRowBrak &last_brak=lst_brak->end()[-1]; \
if (last_brak.y==y) { \
last_brak.end_x=point_x; \
} \
}

int pix_vals_int=*(int*)pix_vals_arr;
WORD *pix_vals_word=(WORD*)pix_vals_arr;
WORD w_part_1=pix_vals_word[0];
WORD w_part_2=pix_vals_word[1];

y+=y_shift;

//PROCESS_BRAK_POINT(x);
//return;

if ((pix_vals_int==-1)||(pix_vals_int==0)||((w_part_1!=0)&&(w_part_2!=0)))
PROCESS_BRAK_POINT(x)
else {
int lst_brak_size=lst_brak->size();

if (*is_start_brak==1) {
//это зона начала брака
if (w_part_1!=0) {
if (lst_brak_size!=0) {
TRowBrak *prior_brak=&lst_brak->end()[-1];
if ((prior_brak->y==y)&&((x-prior_brak->end_x)<=3))
prior_brak->end_x=x+1;
else {
TRowBrak brak;
brak.end_x=x+1;
brak.start_x=x;
brak.y=y;
lst_brak->push_back(brak);
lst_brak_size++;
}
} else {
TRowBrak brak;
brak.end_x=x+1;
brak.start_x=x;
brak.y=y;
lst_brak->push_back(brak);
lst_brak_size++;
}
//закончилась зона брака, т.к. w_part_2==0;
(*is_start_brak)=0;
} else {
//вторая часть четверки байтов не равна нулю
if (lst_brak_size!=0) {
TRowBrak *prior_brak=&lst_brak->end()[-1];
if ((prior_brak->y==y)&&(((x+2)-prior_brak->end_x)<=3))
prior_brak->end_x=x+3;
else {
TRowBrak brak;
brak.end_x=x+3;
brak.start_x=x+2;
brak.y=y;
lst_brak->push_back(brak);
lst_brak_size++;
}
}
}
} else {
//зона конца брака
if (w_part_1!=0) {
if (lst_brak_size!=0) {
TRowBrak *prior_brak=&lst_brak->end()[-1];
if ((prior_brak->y==y)&&((x-prior_brak->end_x)<=3))
prior_brak->end_x=x+1;
else {
TRowBrak brak;
brak.end_x=x+1;
brak.start_x=x;
brak.y=y;
lst_brak->push_back(brak);
lst_brak_size++;
}
}
} else {
//старый брак закончился, новый начался
if (lst_brak_size!=0) {
TRowBrak *prior_brak=&lst_brak->end()[-1];
if ((prior_brak->y==y)&&(((x+2)-prior_brak->end_x)<=3))
prior_brak->end_x=x+3;
else {
TRowBrak brak;
brak.end_x=x+3;
brak.start_x=x+2;
brak.y=y;
lst_brak->push_back(brak);
lst_brak_size++;
}
} else {
TRowBrak brak;
brak.end_x=x+3;
brak.start_x=x+2;
brak.y=y;
lst_brak->push_back(brak);
lst_brak_size++;
}
(*is_start_brak)=1;
}
}
}
}

Вектора создаются динамически через оператор new

Пробовал и перед стартом потоков создавать вектора, и непосредственно внутри обрабатывающего потока.

Тут уже гдето обсуждалось что у них местный аллокатор спинлоками засинхронизирован. Надо переключатсья на CRTшный аллокатор.

Здравствуйте, Аноним, Вы писали:

А>Тут уже гдето обсуждалось что у них местный аллокатор спинлоками засинхронизирован. Надо переключатсья на CRTшный аллокатор.

Я пишу программу под C++Builder 6.0. Соответственно использую его STL.
Что такое STL Port — мне ничего не говорит, тем более CRT STL Port

Подскажите, пожалуйста, что это такое и как его переключить на этот режим.
Заранее спасибо )))

Кароче, заменил vector на обычный длинный массив, быстрее стало — 7 секунд, что не приемлемо ))
Значит, наверное — это, все-таки не STL, а система ...

Здравствуйте, dmitry_filinov, Вы писали:

_>Здравствуйте, Аноним, Вы писали:

А>>Тут уже гдето обсуждалось что у них местный аллокатор спинлоками засинхронизирован. Надо переключатсья на CRTшный аллокатор.

_>Я пишу программу под C++Builder 6.0. Соответственно использую его STL.

Кроме

New STLport C++ Standard Library

[All editions]

C++Builder 6 now includes the STLport multiplatform ANSI C++ Standard Library implementation and uses this library by default. The documentation is included in the main Help directory in HTML format. To learn more, start by looking at the page called bcb6_stlport.html. For more information about STLport, refer to www.stlport.com.

в билдере есть еще и реализация STL

Standard C++ Library is based on the final standard for the C++ language and library ratified in March, 1998, by the American National Standards Institute (ANSI) and the International Standards Organization (ISO)

Спасибо всем !!!!

Я нашел, в чем проблема....

Отключил CodeGuard и все встало на свои места.

И при использовании двух потоков обработки на двух процессорах почти удвоило производительность этого алгоритма.

Здравствуйте, dmitry_filinov, Вы писали:

_>Спасибо всем !!!!

_>Я нашел, в чем проблема....

_>Отключил CodeGuard и все встало на свои места.

_>И при использовании двух потоков обработки на двух процессорах почти удвоило производительность этого алгоритма.

Фигасе... Обычно производительность у релиза меряют, потому что у дебага можно совсем не то намерять, а тут вообще инструментированный код

Извините, челы )))
ну я, просто, отлаживал недавно утечку памяти и забыл, что CodeGuard у меня выставлен.

Здравствуйте, dmitry_filinov, Вы писали:

ты это... форматируй код, пожалуйста, читать же невозможно:

_>перед запуском потоков инициализирую вектора ...
_>typedef struct {
_> int start_x;
_> int end_x;
_> int y;
_>} TRowBrak;
_>typedef std::vector<TRowBrak> TRowBrakVec;

_>.///////////////////////////////////////////////////////////////

_> for (int i=0;i<cpu_core_count;i++) {
_> lp_threads[i].lst_brak=new TRowBrakVec();
_> lp_threads[i].lst_brak->reserve(300000);

_> .....
_> }

_>.///////////////////////////////////////////////////////////////
_>а вот потоковая функция обработки ...
_>void add_brak_to_list(TRowBrakVec *lst_brak, BYTE *pix_vals_arr, BYTE vals_count,
_> int x, int y, BYTE *is_start_brak, int y_shift)
_>{
_> //Sleep(1);
_> //return;

_> #define PROCESS_BRAK_POINT(point_x) \
_> if (*is_start_brak==1) { \
_> int lst_brak_size=lst_brak->size(); \
_> if (lst_brak_size!=0) { \
_> TRowBrak &prior_brak=lst_brak->end()[-1]; \
_> if ((prior_brak.y==y)&&((point_x-prior_brak.end_x)<=3)) \
_> prior_brak.end_x=point_x; \
_> else { \
_> TRowBrak brak; \
_> brak.end_x=0; \
_> brak.start_x=point_x; \
_> brak.y=y; \
_> lst_brak->push_back(brak); \
_> } \
_> } else { \
_> TRowBrak brak; \
_> brak.end_x=0; \
_> brak.start_x=point_x; \
_> brak.y=y;\
_> lst_brak->push_back(brak); \
_> } \
_> } else { \
_> TRowBrak &last_brak=lst_brak->end()[-1]; \
_> if (last_brak.y==y) { \
_> last_brak.end_x=point_x; \
_> } \
_> }

_> int pix_vals_int=*(int*)pix_vals_arr;
_> WORD *pix_vals_word=(WORD*)pix_vals_arr;
_> WORD w_part_1=pix_vals_word[0];
_> WORD w_part_2=pix_vals_word[1];

_> y+=y_shift;

_> //PROCESS_BRAK_POINT(x);
_> //return;

_> if ((pix_vals_int==-1)||(pix_vals_int==0)||((w_part_1!=0)&&(w_part_2!=0)))
_> PROCESS_BRAK_POINT(x)
_> else {
_> int lst_brak_size=lst_brak->size();

_> if (*is_start_brak==1) {
_> //это зона начала брака
_> if (w_part_1!=0) {
_> if (lst_brak_size!=0) {
_> TRowBrak *prior_brak=&lst_brak->end()[-1];
_> if ((prior_brak->y==y)&&((x-prior_brak->end_x)<=3))
_> prior_brak->end_x=x+1;
_> else {
_> TRowBrak brak;
_> brak.end_x=x+1;
_> brak.start_x=x;
_> brak.y=y;
_> lst_brak->push_back(brak);
_> lst_brak_size++;
_> }
_> } else {
_> TRowBrak brak;
_> brak.end_x=x+1;
_> brak.start_x=x;
_> brak.y=y;
_> lst_brak->push_back(brak);
_> lst_brak_size++;
_> }
_> //закончилась зона брака, т.к. w_part_2==0;
_> (*is_start_brak)=0;
_> } else {
_> //вторая часть четверки байтов не равна нулю
_> if (lst_brak_size!=0) {
_> TRowBrak *prior_brak=&lst_brak->end()[-1];
_> if ((prior_brak->y==y)&&(((x+2)-prior_brak->end_x)<=3))
_> prior_brak->end_x=x+3;
_> else {
_> TRowBrak brak;
_> brak.end_x=x+3;
_> brak.start_x=x+2;
_> brak.y=y;
_> lst_brak->push_back(brak);
_> lst_brak_size++;
_> }
_> }
_> }
_> } else {
_> //зона конца брака
_> if (w_part_1!=0) {
_> if (lst_brak_size!=0) {
_> TRowBrak *prior_brak=&lst_brak->end()[-1];
_> if ((prior_brak->y==y)&&((x-prior_brak->end_x)<=3))
_> prior_brak->end_x=x+1;
_> else {
_> TRowBrak brak;
_> brak.end_x=x+1;
_> brak.start_x=x;
_> brak.y=y;
_> lst_brak->push_back(brak);
_> lst_brak_size++;
_> }
_> }
_> } else {
_> //старый брак закончился, новый начался
_> if (lst_brak_size!=0) {
_> TRowBrak *prior_brak=&lst_brak->end()[-1];
_> if ((prior_brak->y==y)&&(((x+2)-prior_brak->end_x)<=3))
_> prior_brak->end_x=x+3;
_> else {
_> TRowBrak brak;
_> brak.end_x=x+3;
_> brak.start_x=x+2;
_> brak.y=y;
_> lst_brak->push_back(brak);
_> lst_brak_size++;
_> }
_> } else {
_> TRowBrak brak;
_> brak.end_x=x+3;
_> brak.start_x=x+2;
_> brak.y=y;
_> lst_brak->push_back(brak);
_> lst_brak_size++;
_> }
_> (*is_start_brak)=1;
_> }
_> }
_> }
_>}

You will always get what you always got
  If you always do  what you always did