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

#include "TableMapper.h"
#include <assert.h>
#include <process.h>
#include <iostream>
#include <stdio.h>

void TableMapper::Map()
{
	if (m_pTable == NULL)
		return;
	m_map.clear();
	HRecCItor beginIt = m_pTable->GetBeginCIterator();
	HRecCItor endIt   = m_pTable->GetEndCIterator();
	for (; beginIt != endIt; ++beginIt)
		MapItem(*beginIt);
	if (m_pShuffler)
		m_pShuffler->Shuffle(&m_map);
}

void TableMapper::MapItem(const HREC pRec)
{
	const SampleItem* pSample = (const SampleItem*)pRec;
	if (pSample == NULL)
		return;
	m_map.insert(std::make_pair(pSample->price, pSample->name));
#ifdef _DEBUG_MSG
	wchar_t buf[1024] = {0,};
	swprintf_s(buf, L"Insert to map: %d:%s\n", pSample->price, pSample->name.c_str());
	std::wcout<<buf;
#endif
}

void TableShuffler::Shuffle(const SampleMap* map)
{
	assert(map);
	EnterCriticalSection(&m_cs);
	for (SampleMapCItor it = map->begin(); it != map->end(); ++it)
	{
		m_map.insert(std::make_pair(it->first, it->second));
	}
	++m_count;
	LeaveCriticalSection(&m_cs);
}

void TableReducer::Reduce(const SampleMap& map)
{
	for (SampleMapCItor it = map.begin(); it != map.end(); ++it)
		Count(it->first);
}

void TableReducer::Count(int number)
{
	CountMapItor countIt = m_countMap.find(number);
	if (countIt != m_countMap.end())
	{
		countIt->second += 1;
		return;
	}
	m_countMap.insert(std::make_pair(number, 1));
}

void TableReducer::PrintOutput()
{
	for (CountMapItor it = m_countMap.begin(); it != m_countMap.end(); ++it)
		std::wcout<<L"price "<<it->first<<L" counts "<<it->second<<L"\n";
}

#ifdef UNITTEST
#include "TableLoader.h"
SUITE(Mapper)
{

struct SampleMapper
{
	SampleTable	table1;
	SampleTable	table2;
	TableLoader	loader;
	TableMapper	mapper1;
	TableMapper	mapper2;
	TableShuffler	shuffler;

	SampleMapper()
	{
		table1.SetFileName(L"Sample.xml");
		table2.SetFileName(L"Sample2.xml");
		loader.Load(&table1);
		loader.Load(&table2);
		mapper1.SetTable(&table1);
		mapper2.SetTable(&table2);
		mapper1.SetShuffler(&shuffler);
		mapper2.SetShuffler(&shuffler);
	}
};

void MapThread(void* _mapper)
{
	TableMapper *mapper = (TableMapper*)_mapper;
	mapper->Map();
}

TEST_FIXTURE(SampleMapper, MapShuffleReduce)
{
	_beginthread(MapThread, 0, &mapper1);
	_beginthread(MapThread, 0, &mapper2);
	while (shuffler.GetShuffleCount()<2)
		Sleep(1000);
	TableReducer	reducer;
	reducer.Reduce(shuffler.GetMap());
#ifdef _DEBUG_MSG
	reducer.PrintOutput();
#endif
}

}
#endif // UNITTEST