andre/CPipe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Thread test works!

Browse Source
This commit is contained in:
Andre Heber
2016-03-05 23:57:18 +01:00
parent ccc5014cb8
commit ae55b0b4f8
14 changed files with 5781 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
.cproject Normal file
View File

@ -0,0 +1,44 @@
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<?fileVersion 4.0.0?><cproject storage_type_id="org.eclipse.cdt.core.XmlProjectDescriptionStorage">
<storageModule moduleId="org.eclipse.cdt.core.settings">
<cconfiguration id="0.598556019">
<storageModule buildSystemId="org.eclipse.cdt.managedbuilder.core.configurationDataProvider" id="0.598556019" moduleId="org.eclipse.cdt.core.settings" name="Default">
<externalSettings/>
<extensions>
<extension id="org.eclipse.cdt.ui.Regex Error Parser" point="org.eclipse.cdt.core.ErrorParser"/>
</extensions>
</storageModule>
<storageModule moduleId="cdtBuildSystem" version="4.0.0">
<configuration buildProperties="" description="" errorParsers="org.eclipse.cdt.ui.Regex Error Parser" id="0.598556019" name="Default" parent="org.eclipse.cdt.build.core.prefbase.cfg">
<folderInfo id="0.598556019." name="/" resourcePath="">
<toolChain errorParsers="" id="org.eclipse.cdt.build.core.prefbase.toolchain.1235403699" name="No ToolChain" resourceTypeBasedDiscovery="false" superClass="org.eclipse.cdt.build.core.prefbase.toolchain">
<targetPlatform id="org.eclipse.cdt.build.core.prefbase.toolchain.1235403699.413443094" name=""/>
<builder arguments="-f &quot;../Tool/toolchain/opt/compiler.opt&quot; --core=tc1.3.1 --cpu=tc1782 --source -f &quot;${selected_resource_loc}/../../tmp/Includes.opt&quot; -f &quot;${selected_resource_loc}/../../opt/compiler.opt&quot; -o &quot;${selected_resource_loc}/../../tmp/trash&quot; &quot;${selected_resource_loc}&quot;" command="&quot;d:/tools/Tasking/TriCore/3.2r3/ctc/bin/ctc.exe&quot;" enableCleanBuild="false" errorParsers="" id="org.eclipse.cdt.build.core.settings.default.builder.1123914364" incrementalBuildTarget="" keepEnvironmentInBuildfile="false" managedBuildOn="false" name="Gnu Make Builder" superClass="org.eclipse.cdt.build.core.settings.default.builder"/>
<tool errorParsers="" id="org.eclipse.cdt.build.core.settings.holder.libs.1396912558" name="holder for library settings" superClass="org.eclipse.cdt.build.core.settings.holder.libs"/>
<tool errorParsers="" id="org.eclipse.cdt.build.core.settings.holder.1043027053" name="Assembly" superClass="org.eclipse.cdt.build.core.settings.holder">
<inputType id="org.eclipse.cdt.build.core.settings.holder.inType.1089353706" languageId="org.eclipse.cdt.core.assembly" languageName="Assembly" sourceContentType="org.eclipse.cdt.core.asmSource" superClass="org.eclipse.cdt.build.core.settings.holder.inType"/>
</tool>
<tool errorParsers="" id="org.eclipse.cdt.build.core.settings.holder.2057322761" name="GNU C++" superClass="org.eclipse.cdt.build.core.settings.holder">
<inputType id="org.eclipse.cdt.build.core.settings.holder.inType.1478445541" languageId="org.eclipse.cdt.core.g++" languageName="GNU C++" sourceContentType="org.eclipse.cdt.core.cxxSource,org.eclipse.cdt.core.cxxHeader" superClass="org.eclipse.cdt.build.core.settings.holder.inType"/>
</tool>
<tool errorParsers="" id="org.eclipse.cdt.build.core.settings.holder.816818701" name="GNU C" superClass="org.eclipse.cdt.build.core.settings.holder">
<inputType id="org.eclipse.cdt.build.core.settings.holder.inType.1814575165" languageId="org.eclipse.cdt.core.gcc" languageName="GNU C" sourceContentType="org.eclipse.cdt.core.cSource,org.eclipse.cdt.core.cHeader" superClass="org.eclipse.cdt.build.core.settings.holder.inType"/>
</tool>
</toolChain>
</folderInfo>
</configuration>
</storageModule>
<storageModule moduleId="org.eclipse.cdt.core.externalSettings"/>
</cconfiguration>
</storageModule>
<storageModule moduleId="cdtBuildSystem" version="4.0.0">
<project id="Pipe.null.2005735511" name="Pipe"/>
</storageModule>
<storageModule moduleId="scannerConfiguration">
<autodiscovery enabled="true" problemReportingEnabled="true" selectedProfileId=""/>
<scannerConfigBuildInfo instanceId="0.598556019">
<autodiscovery enabled="true" problemReportingEnabled="true" selectedProfileId=""/>
</scannerConfigBuildInfo>
</storageModule>
<storageModule moduleId="org.eclipse.cdt.core.LanguageSettingsProviders"/>
</cproject>

26
.project Normal file
View File

@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<projectDescription>
<name>Pipe</name>
<comment></comment>
<projects>
</projects>
<buildSpec>
<buildCommand>
<name>org.eclipse.cdt.managedbuilder.core.genmakebuilder</name>
<triggers>full,incremental,</triggers>
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.eclipse.cdt.managedbuilder.core.ScannerConfigBuilder</name>
<triggers>full,incremental,</triggers>
<arguments>
</arguments>
</buildCommand>
</buildSpec>
<natures>
<nature>org.eclipse.cdt.core.cnature</nature>
<nature>org.eclipse.cdt.managedbuilder.core.managedBuildNature</nature>
<nature>org.eclipse.cdt.managedbuilder.core.ScannerConfigNature</nature>
</natures>
</projectDescription>

1
Pipe/Pipe.vcxproj
View File

@ -114,6 +114,7 @@
<ClCompile Include="threads.c" />
</ItemGroup>
<ItemGroup>
<ClInclude Include="connector.h" />
<ClInclude Include="pipe.h" />
<ClInclude Include="ringbuffer.h" />
</ItemGroup>

3
Pipe/Pipe.vcxproj.filters
View File

@ -29,5 +29,8 @@
<ClInclude Include="ringbuffer.h">
<Filter>Header Files</Filter>
</ClInclude>
<ClInclude Include="connector.h">
<Filter>Header Files</Filter>
</ClInclude>
</ItemGroup>
</Project>

0
Pipe/buffer.txt Normal file
View File

100
Pipe/buffer1.txt Normal file
View File

@ -0,0 +1,100 @@
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
0
0
0
0
0
0
0
0
0
0
0
0
0

100
Pipe/buffer2.txt Normal file
View File

@ -0,0 +1,100 @@
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
0

95
Pipe/connector.h Normal file
View File

@ -0,0 +1,95 @@
#ifndef CONNECTION_H_
#define CONNECTION_H_
#include <stdint.h>
#include "ringbuffer.h"
/* microsoft specific */
#ifdef _MSC_VER
#define inline __inline
#endif
/* A connection is an array of pointers, pointing to ringbuffers. */
typedef struct
{
ringbuffer_t ** connection;
uint32_t connection_count;
uint32_t connection_max;
} connector_t;
static inline uint8_t Connector_IsFull(connector_t * const connector)
{
for (uint32_t i = 0; i < connector->connection_count; i++)
{
if (RingBuffer_IsFull(connector->connection[i]))
return 1;
}
return 0;
}
static inline uint8_t Connector_IsEmpty(connector_t * const connector)
{
for (uint32_t i = 0; i < connector->connection_count; i++)
{
if (!RingBuffer_IsEmpty(connector->connection[i]))
return 0;
}
return 1;
}
static inline uint8_t Connector_IsFilled(connector_t * const connector)
{
return !Connector_IsEmpty(connector);
}
static inline uint32_t Connector_Read(connector_t * const connector)
{
uint32_t i = 0;
while ( RingBuffer_IsEmpty(connector->connection[i]) && (i < connector->connection_count) )
i++;
if (i == connector->connection_count)
return 0;
return RingBuffer_Read(connector->connection[i]);
}
static inline void Connector_Write(connector_t * const connector, uint32_t element)
{
for (uint32_t i = 0; i < connector->connection_count; i++)
RingBuffer_Write(connector->connection[i], element);
}
static inline void Connector_Insert(connector_t * const connector, uint32_t ring_buffer_index, uint32_t element)
{
RingBuffer_Write(connector->connection[ring_buffer_index], element);
}
static inline void Connector_Add(connector_t * const connector, ringbuffer_t * ring_buffer)
{
if (connector->connection_count < connector->connection_max)
{
connector->connection[connector->connection_count] = ring_buffer;
connector->connection_count++;
}
}
static inline void Connector_Init(connector_t * const connector, ringbuffer_t *buffer[], uint32_t buffer_size)
{
connector->connection = &buffer[0];
connector->connection_count = 0;
connector->connection_max = buffer_size;
}
#define Concat2(a, b) a ## b
#define Concat(a, b) Concat2(a, b)
/*
Macro for the creation of a connector.
*/
#define Connector_Create(name, size) \
ringbuffer_t *Concat(name, Concat(_buffer, __LINE__))[size]; \
connector_t name; \
Connector_Init(&name, Concat(name, Concat(_buffer, __LINE__)), size); \
#endif

67
Pipe/main.c
View File

@ -4,6 +4,7 @@
#include <windows.h>
#include "ringbuffer.h"
#include "connector.h"
#include "pipe.h"
#ifdef _MSC_VER
@ -13,7 +14,7 @@
/* Integrate every element of the signal. */
void increment(pipe_t * const p)
{
while (Pipe_isFilled(p))
while (Pipe_IsFilled(p))
{
uint32_t item = Pipe_Read(p);
item++;
@ -24,7 +25,7 @@ void increment(pipe_t * const p)
/* Square every element of the signal. */
void square(pipe_t * const p)
{
while (Pipe_isFilled(p))
while (Pipe_IsFilled(p))
{
uint32_t item = Pipe_Read(p);
item = item * item;
@ -37,7 +38,7 @@ void integrate(pipe_t * const pipe)
{
uint32_t state = *((uint32_t*)pipe->state);
while (Pipe_isFilled(pipe))
while (Pipe_IsFilled(pipe))
{
uint32_t item = Pipe_Read(pipe);
state = state + item;
@ -52,7 +53,7 @@ void sum(pipe_t * const pipe)
{
uint32_t sum = 0;
while (Pipe_isFilled(pipe))
while (Pipe_IsFilled(pipe))
sum += Pipe_Read(pipe);
Pipe_Write(pipe, sum);
}
@ -64,7 +65,7 @@ void average(pipe_t * const pipe)
uint32_t element_counter = 0;
uint32_t average = 0;
while (Pipe_isFilled(pipe))
while (Pipe_IsFilled(pipe))
{
sum += Pipe_Read(pipe);
element_counter++;
@ -77,27 +78,30 @@ void average(pipe_t * const pipe)
void print(pipe_t * const pipe)
{
printf("\nOutput:\n");
while (Pipe_isFilled(pipe))
while (Pipe_IsFilled(pipe))
printf("%d\n", Pipe_Read(pipe));
}
/* Logging function. Set by user. */
void log(pipe_t * const source, pipe_t * const target, uint32_t element)
{
if (Pipe_isFull(target))
printf("Error: Pipe %s is full!\n", target->name);
//if (Pipe_IsFull(target))
// printf("Error: Pipe %s is full!\n", target->name);
if (source->state == NULL && target->state == NULL)
printf("%s -> %d -> %s\n", source->name, element, target->name);
else if (source->state != NULL && target->state != NULL)
printf("%s(%d) -> %d -> %s(%d)\n", source->name, *((uint32_t*)source->state), element, target->name, *((uint32_t*)target->state));
else if (source->state != NULL)
printf("%s(%d) -> %d -> %s\n", source->name, *((uint32_t*)source->state), element, target->name);
else
printf("%s -> %d -> %s(%d)\n", source->name, element, target->name, *((uint32_t*)target->state));
//if (source->state == NULL && target->state == NULL)
// printf("%s -> %d -> %s\n", source->name, element, target->name);
//else if (source->state != NULL && target->state != NULL)
// printf("%s(%d) -> %d -> %s(%d)\n", source->name, *((uint32_t*)source->state), element, target->name, *((uint32_t*)target->state));
//else if (source->state != NULL)
// printf("%s(%d) -> %d -> %s\n", source->name, *((uint32_t*)source->state), element, target->name);
//else
// printf("%s -> %d -> %s(%d)\n", source->name, element, target->name, *((uint32_t*)target->state));
printf("%s: %d\n", source->name, element);
}
extern void threads(void);
extern void threads(uint32_t loops);
int main(int argc, char *argv[])
{
uint32_t counter = 0;
@ -105,19 +109,22 @@ int main(int argc, char *argv[])
if (argc == 1)
{
/* Create pipes and connect them */
Pipe_Create(increment_pipe, 4, 1, NULL, log);
Pipe_Create(square_pipe, 4, 1, NULL, log);
Pipe_Create(integrate_pipe, 8, 2, &counter, log);
Pipe_Create(sum_pipe, 8, 2, NULL, log);
Pipe_Create(average_pipe, 8, 2, NULL, log);
Pipe_Create(print_pipe, 4, 2, NULL, log);
Pipe_Create(increment_pipe, 1, 1, NULL, log);
Pipe_Create(square_pipe, 1, 1, NULL, log);
Pipe_Create(integrate_pipe, 2, 2, &counter, log);
Pipe_Create(sum_pipe, 1, 1, NULL, log);
Pipe_Create(average_pipe, 1, 1, NULL, log);
Pipe_Create(print_pipe, 2, 1, NULL, log);
Pipe_Connect(&increment_pipe, &integrate_pipe);
Pipe_Connect(&square_pipe, &integrate_pipe);
Pipe_Connect(&integrate_pipe, &sum_pipe);
Pipe_Connect(&integrate_pipe, &average_pipe);
Pipe_Connect(&sum_pipe, &print_pipe);
Pipe_Connect(&average_pipe, &print_pipe);
Pipe_CreateInputBuffer(increment_pipe, 4);
Pipe_CreateInputBuffer(square_pipe, 4);
Pipe_CreateConnection(increment_pipe, integrate_pipe, 4);
Pipe_CreateConnection(square_pipe, integrate_pipe, 4);
Pipe_CreateConnection(integrate_pipe, sum_pipe, 8);
Pipe_CreateConnection(integrate_pipe, average_pipe, 8);
Pipe_CreateConnection(sum_pipe, print_pipe, 4);
Pipe_CreateConnection(average_pipe, print_pipe, 4);
/* Create Input */
Pipe_Insert(&increment_pipe, 1);
@ -140,7 +147,7 @@ int main(int argc, char *argv[])
}
else
{
threads();
threads(atoi(argv[1]));
}
return 0;

78
Pipe/pipe.h
View File

@ -3,6 +3,7 @@
#include <stdint.h>
#include "ringbuffer.h"
#include "connector.h"
/* microsoft specific */
#ifdef _MSC_VER
@ -11,11 +12,9 @@
typedef struct pipe_tt
{
ringbuffer_t * input;
connector_t * input_connector;
connector_t * output_connector;
void * state;
struct pipe_tt ** connection;
uint32_t connection_count;
uint32_t connection_max;
char * name;
void(*log_function)(struct pipe_tt * from, struct pipe_tt * to, uint32_t elem);
} pipe_t;
@ -31,7 +30,7 @@ So the signal can inserted to the pipe system.
*/
static inline void Pipe_Insert(pipe_t * const pipe, uint32_t element)
{
RingBuffer_Write(pipe->input, element);
Connector_Insert(pipe->input_connector, 0, element);
}
/*
@ -40,7 +39,7 @@ Used inside functions of the pipe system.
*/
static inline uint32_t Pipe_Read(pipe_t * pipe)
{
return RingBuffer_Read(pipe->input);
return Connector_Read(pipe->input_connector);
}
/*
@ -49,29 +48,26 @@ Used inside functions of the pipe system.
*/
static inline void Pipe_Write(pipe_t * pipe, uint32_t element)
{
for (uint8_t i = 0; i < pipe->connection_count; i++)
{
pipe->log_function(pipe, pipe->connection[i], element);
RingBuffer_Write(pipe->connection[i]->input, element);
}
pipe->log_function(pipe, NULL, element);
Connector_Write(pipe->output_connector, element);
}
/*
Check if the pipe contents elements.
Used inside functions of the pipe system.
*/
static inline uint8_t Pipe_isFilled(pipe_t * pipe)
static inline uint8_t Pipe_IsFilled(pipe_t * pipe)
{
return RingBuffer_IsFilled(pipe->input);
return Connector_IsFilled(pipe->input_connector);
}
/*
Check if the pipe has no place left for new elements.
Usefull for the logging.
*/
static inline uint8_t Pipe_isFull(const pipe_t * pipe)
static inline uint8_t Pipe_IsFull(const pipe_t * pipe)
{
return RingBuffer_IsFull(pipe->input);
return Connector_IsFull(pipe->input_connector);
}
/*
@ -80,7 +76,7 @@ Shall be set by user.
*/
static inline void Pipe_Log(pipe_t * const source, pipe_t * const target, uint32_t element)
{
if (Pipe_isFull(target))
if (Pipe_IsFull(target))
printf("Error: Pipe %s is full!\n", target->name);
if (source->state == NULL && target->state == NULL)
@ -100,7 +96,6 @@ static inline void Pipe_Log(pipe_t * const source, pipe_t * const target, uint32
#define Concat2(a, b) a ## b
#define Concat(a, b) Concat2(a, b)
#define SizeOfArray(arg) ( sizeof(arg) / sizeof(arg[0]) )
/*
Macro for the creation of a pipe.
@ -111,13 +106,11 @@ arg_conn_count is the number of outgoing connections.
arg_state is the given state, which can be used in the function.
arg_log is the log function, called when an element is sent.
*/
#define Pipe_Create(arg_name, arg_size, arg_conn_count, arg_state, arg_log) \
static uint32_t Concat(arg_name, Concat(_buffer, __LINE__))[arg_size]; \
ringbuffer_t arg_name ## _rb; \
RingBuffer_InitFromArray(&arg_name ## _rb, Concat(arg_name, Concat(_buffer, __LINE__)), arg_size); \
pipe_t arg_name; \
static pipe_t *Concat(arg_name, Concat(_connection_buffer, __LINE__))[arg_conn_count]; \
Pipe_Init(&arg_name, &arg_name ## _rb, Concat(arg_name, Concat(_connection_buffer, __LINE__)), arg_conn_count, arg_state, #arg_name, arg_log)
#define Pipe_Create(name, input_connection_count, output_connection_count, state, log) \
static Connector_Create(Concat(name, Concat(_input_connector, __LINE__)), input_connection_count); \
static Connector_Create(Concat(name, Concat(_output_connector, __LINE__)), output_connection_count); \
pipe_t name; \
Pipe_Init(&name, &Concat(name, Concat(_input_connector, __LINE__)), &Concat(name, Concat(_output_connector, __LINE__)), state, #name, log)
/*
Initializes a pipe.
@ -127,23 +120,16 @@ A Name and a logging function are usefull to track the dataflow.
*/
static inline void Pipe_Init(
pipe_t * const pipe,
ringbuffer_t * const input,
pipe_t ** pipe_connections,
uint32_t connection_max,
connector_t * input_connector,
connector_t * output_connector,
void * state,
char * const name,
void(*log_function)(struct pipe_tt * source, struct pipe_tt * target, uint32_t element)
)
{
pipe->input = input;
pipe->input_connector = input_connector;
pipe->output_connector = output_connector;
pipe->state = state;
pipe->connection = pipe_connections;
for (uint8_t i = 0; i < connection_max; i++)
pipe->connection[i] = NULL;
pipe->connection_count = 0;
pipe->connection_max = connection_max;
pipe->name = name;
if (log_function == NULL)
@ -153,13 +139,23 @@ static inline void Pipe_Init(
}
/* Connect two pipes. Pipe a sends elements to pipe b. */
static inline void Pipe_Connect(pipe_t * const source, pipe_t * const target)
static inline void Pipe_Connect(pipe_t * const source, pipe_t * const target, ringbuffer_t * ring_buffer)
{
if (source->connection_count < source->connection_max)
{
source->connection[source->connection_count] = target;
source->connection_count++;
}
Connector_Add(source->output_connector, ring_buffer);
Connector_Add(target->input_connector, ring_buffer);
}
#define Pipe_CreateConnection(source, target, size) \
RingBuffer_Create(Concat(connector_ring_buffer, __LINE__), size); \
Pipe_Connect(&source, &target, &Concat(connector_ring_buffer, __LINE__))
static inline void Pipe_AddInputBuffer(pipe_t * const pipe, ringbuffer_t * input_buffer)
{
Connector_Add(pipe->input_connector, input_buffer);
}
#define Pipe_CreateInputBuffer(pipe_name, size) \
RingBuffer_Create(Concat(connector_ring_buffer, __LINE__), size); \
Pipe_AddInputBuffer(&pipe_name, &Concat(connector_ring_buffer, __LINE__))
#endif

15
Pipe/ringbuffer.h
View File

@ -15,10 +15,13 @@ typedef struct {
} ringbuffer_t;
/* Use Array as RingBuffer */
static inline void RingBuffer_InitFromArray(ringbuffer_t * const ring_buffer, uint32_t * const array, uint32_t size)
static inline void RingBuffer_Init(ringbuffer_t * const ring_buffer, uint32_t * const array, uint32_t size)
{
ring_buffer->start = ring_buffer->reader = ring_buffer->writer = &array[0];
ring_buffer->end = &array[0] + size - 1;
for (uint32_t i = 0; i < size; i++)
array[i] = 0;
}
static inline uint32_t * RingBuffer_NextAddress(ringbuffer_t * const ring_buffer, uint32_t * const pointer)
@ -66,11 +69,19 @@ static inline uint32_t RingBuffer_Read(ringbuffer_t * const ring_buffer)
if (!RingBuffer_IsEmpty(ring_buffer))
{
element = *(ring_buffer->reader);
*(ring_buffer->reader) = 5;
*(ring_buffer->reader) = 30;
ring_buffer->reader = RingBuffer_NextAddress(ring_buffer, ring_buffer->reader);
}
return element;
}
/*
Macro for the creation of a ring buffer.
*/
#define RingBuffer_Create(name, size) \
uint32_t Concat(name, Concat(_buffer, __LINE__))[size]; \
ringbuffer_t name; \
RingBuffer_Init(&name, Concat(name, Concat(_buffer, __LINE__)), size) \
#endif

2235
Pipe/thread1.txt Normal file
View File

File diff suppressed because it is too large Load Diff

2852
Pipe/thread2.txt Normal file
View File

File diff suppressed because it is too large Load Diff

349
Pipe/threads.c
View File

@ -6,103 +6,32 @@
#include "ringbuffer.h"
#include "pipe.h"
#define NUMBER_OF_ELEMENTS 1000
/* Integrate every element of the signal. */
static void increment(pipe_t * const p)
static void increment(pipe_t * const pipe)
{
while (Pipe_isFilled(p))
uint8_t check = 1;
while (Pipe_IsFilled(pipe))
{
uint32_t item = Pipe_Read(p);
item++;
Pipe_Write(p, item);
uint32_t item = Pipe_Read(pipe);
// do something
Pipe_Write(pipe, item);
}
}
/* Print the signal. */
static void write_to_file1(pipe_t * const pipe)
{
HANDLE hFile;
char DataBuffer[128];
DWORD dwBytesWritten = 0;
hFile = CreateFile("thread1.txt", GENERIC_WRITE, 0, NULL, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
while (1) {
while (Pipe_isFilled(pipe))
{
sprintf_s(DataBuffer, 128, "%d\r\n", Pipe_Read(pipe));
WriteFile(hFile, DataBuffer, (DWORD)strlen(DataBuffer), &dwBytesWritten, NULL);
}
}
}
static void write_to_file2(pipe_t * const pipe)
{
HANDLE hFile;
char DataBuffer[128];
DWORD dwBytesWritten = 0;
hFile = CreateFile("thread2.txt", GENERIC_WRITE, 0, NULL, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
while (1) {
while (Pipe_isFilled(pipe))
{
sprintf_s(DataBuffer, 128, "%d\r\n", Pipe_Read(pipe));
WriteFile(hFile, DataBuffer, (DWORD)strlen(DataBuffer), &dwBytesWritten, NULL);
}
}
}
DWORD WINAPI incrementThread1(LPVOID lpParam)
{
pipe_t * pipe = (pipe_t *)lpParam;
LARGE_INTEGER time;
while (1)
{
/* Generate input */
QueryPerformanceCounter(&time);
Pipe_Insert(pipe, (uint32_t)time.u.LowPart);
QueryPerformanceCounter(&time);
Pipe_Insert(pipe, (uint32_t)time.u.LowPart);
QueryPerformanceCounter(&time);
Pipe_Insert(pipe, (uint32_t)time.u.LowPart);
increment(pipe);
}
return 0;
}
DWORD WINAPI incrementThread2(LPVOID lpParam)
{
pipe_t * pipe = (pipe_t *)lpParam;
LARGE_INTEGER time;
while (1)
{
/* Generate input */
QueryPerformanceCounter(&time);
Pipe_Insert(pipe, (uint32_t)time.u.LowPart);
QueryPerformanceCounter(&time);
Pipe_Insert(pipe, (uint32_t)time.u.LowPart);
QueryPerformanceCounter(&time);
Pipe_Insert(pipe, (uint32_t)time.u.LowPart);
increment(pipe);
}
return 0;
}
DWORD WINAPI incrementThread3(LPVOID lpParam)
DWORD WINAPI Thread1(LPVOID lpParam)
{
pipe_t * pipe = (pipe_t *)lpParam;
while (1)
{
increment(pipe);
}
/* Generate input */
for (int i = 0; i < NUMBER_OF_ELEMENTS; i++)
Pipe_Insert(pipe, 42);
increment(pipe);
Pipe_Write(pipe, UINT32_MAX);
return 0;
}
@ -110,50 +39,242 @@ DWORD WINAPI incrementThread3(LPVOID lpParam)
DWORD WINAPI Thread2(LPVOID lpParam)
{
pipe_t * pipe = (pipe_t *)lpParam;
write_to_file2(pipe);
/* Generate input */
for (int i = 0; i < NUMBER_OF_ELEMENTS; i++)
Pipe_Insert(pipe, 23);
increment(pipe);
Pipe_Write(pipe, UINT32_MAX);
return 0;
}
DWORD WINAPI Thread1(LPVOID lpParam)
DWORD WINAPI Thread3(LPVOID lpParam)
{
pipe_t * pipe = (pipe_t *)lpParam;
write_to_file1(pipe);
uint32_t item = 0;
uint8_t check = 1;
uint8_t symbol_counter = 0;
while (symbol_counter < 2)
{
while (Pipe_IsFilled(pipe))
{
item = Pipe_Read(pipe);
if (item == UINT32_MAX)
{
symbol_counter++;
break;
}
if (item != 42 && item != 23)
printf("failed\n");
Pipe_Write(pipe, item);
}
}
Pipe_Write(pipe, UINT32_MAX);
return 0;
}
static void log(pipe_t * const source, pipe_t * const target, uint32_t element) {}
DWORD WINAPI Thread4(LPVOID lpParam)
{
pipe_t * pipe = (pipe_t *)lpParam;
uint32_t item = 0;
uint8_t check = 1;
void threads(void)
while (item != UINT32_MAX)
{
while (Pipe_IsFilled(pipe))
{
item = Pipe_Read(pipe);
if (item == UINT32_MAX)
break;
if (item != 42 && item != 23)
printf("failed\n");
Pipe_Write(pipe, item);
}
}
Pipe_Write(pipe, UINT32_MAX);
return 0;
}
DWORD WINAPI Thread5(LPVOID lpParam)
{
pipe_t * pipe = (pipe_t *)lpParam;
uint32_t item = 0;
uint8_t check = 1;
while (item != UINT32_MAX)
{
while (Pipe_IsFilled(pipe))
{
item = Pipe_Read(pipe);
if (item == 0 || item > 102)
{
check = 0;
}
if (item == UINT32_MAX)
break;
Pipe_Write(pipe, item);
}
}
Pipe_Write(pipe, UINT32_MAX);
return 0;
}
void Run_Threads(pipe_t * pipe1, pipe_t * pipe2, pipe_t * pipe3, pipe_t * pipe4, pipe_t * pipe5)
{
/* Create Threads */
HANDLE threadHandles[5];
threadHandles[0] = CreateThread(NULL, 0, Thread1, pipe1, 0, NULL);
threadHandles[1] = CreateThread(NULL, 0, Thread2, pipe2, 0, NULL);
threadHandles[2] = CreateThread(NULL, 0, Thread3, pipe3, 0, NULL);
threadHandles[3] = CreateThread(NULL, 0, Thread4, pipe4, 0, NULL);
threadHandles[4] = CreateThread(NULL, 0, Thread5, pipe5, 0, NULL);
WaitForMultipleObjects(5, threadHandles, TRUE, INFINITE);
CloseHandle(threadHandles[0]);
CloseHandle(threadHandles[1]);
CloseHandle(threadHandles[2]);
CloseHandle(threadHandles[3]);
CloseHandle(threadHandles[4]);
}
uint8_t ringbuffer_valid_and_equal(ringbuffer_t * rb1, ringbuffer_t * rb2)
{
//char DataBuffer[128];
//DWORD dwBytesWritten;
uint32_t element1 = 0;
uint32_t element2 = 0;
uint32_t element_42_counter = 0;
uint32_t element_23_counter = 0;
while (1)
{
element1 = RingBuffer_Read(rb1);
element2 = RingBuffer_Read(rb2);
if (element1 != element2)
printf("Elements unequal!\n");
if (element1 == UINT32_MAX)
break;
if (element1 == 42)
{
element_42_counter++;
//sprintf_s(DataBuffer, 128, "%u - %u\r\n", element1, element_42_counter);
}
if (element1 == 23)
{
element_23_counter++;
//sprintf_s(DataBuffer, 128, "%u - %u\r\n", element1, element_23_counter);
}
//WriteFile(h, DataBuffer, (DWORD)strlen(DataBuffer), &dwBytesWritten, NULL);
}
if ((element_42_counter != element_23_counter) || (element_42_counter != NUMBER_OF_ELEMENTS))
{
printf("error!\n");
return 0;
}
return 1;
}
static void log(pipe_t * const source, pipe_t * const target, uint32_t element)
{
/*wchar_t text[64];
swprintf_s(text, 64, L"%s: %d\n", source->name, element);
OutputDebugStringW(text);*/
}
void threads(uint32_t loops)
{
/* Create pipes and connect them */
Pipe_Create(increment_pipe1, 4, 1, NULL, log);
Pipe_Create(increment_pipe2, 4, 1, NULL, log);
Pipe_Create(increment_pipe3, 8, 2, NULL, log);
Pipe_Create(write_to_file1_pipe, 8, 1, NULL, log);
Pipe_Create(write_to_file2_pipe, 8, 1, NULL, log);
Pipe_Create(increment_pipe1, 1, 1, NULL, log);
Pipe_Create(increment_pipe2, 1, 1, NULL, log);
Pipe_Create(increment_pipe3, 2, 2, NULL, log);
Pipe_Create(increment_pipe4, 1, 1, NULL, log);
Pipe_Create(increment_pipe5, 1, 1, NULL, log);
Pipe_Create(check_pipe1, 1, 1, NULL, log);
Pipe_Create(check_pipe2, 1, 1, NULL, log);
Pipe_Connect(&increment_pipe1, &increment_pipe3);
Pipe_Connect(&increment_pipe2, &increment_pipe3);
Pipe_Connect(&increment_pipe3, &write_to_file1_pipe);
Pipe_Connect(&increment_pipe3, &write_to_file2_pipe);
Pipe_CreateInputBuffer(increment_pipe1, NUMBER_OF_ELEMENTS+2);
Pipe_CreateInputBuffer(increment_pipe2, NUMBER_OF_ELEMENTS+2);
/* Create Threads */
CreateThread(NULL, 0, incrementThread1, &increment_pipe1, 0, NULL);
CreateThread(NULL, 0, incrementThread2, &increment_pipe2, 0, NULL);
CreateThread(NULL, 0, incrementThread3, &increment_pipe3, 0, NULL);
Pipe_CreateConnection(increment_pipe1, increment_pipe3, NUMBER_OF_ELEMENTS+2);
Pipe_CreateConnection(increment_pipe2, increment_pipe3, NUMBER_OF_ELEMENTS+2);
Pipe_CreateConnection(increment_pipe3, increment_pipe4, NUMBER_OF_ELEMENTS * 2 + 2);
Pipe_CreateConnection(increment_pipe3, increment_pipe5, NUMBER_OF_ELEMENTS * 2 + 2);
Pipe_CreateConnection(increment_pipe4, check_pipe1, NUMBER_OF_ELEMENTS * 2 + 2);
Pipe_CreateConnection(increment_pipe5, check_pipe2, NUMBER_OF_ELEMENTS * 2 + 2);
HANDLE thread1Handle, thread2Handle;
//HANDLE hFile;
//hFile = CreateFile("buffer.txt", GENERIC_WRITE, 0, NULL, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
thread1Handle = CreateThread(NULL, 0, Thread1, &write_to_file1_pipe, 0, NULL);
thread2Handle = CreateThread(NULL, 0, Thread2, &write_to_file2_pipe, 0, NULL);
SYSTEMTIME local_time;
GetLocalTime(&local_time);
HANDLE threadHandles[] = { thread1Handle, thread2Handle };
WaitForMultipleObjects(2, threadHandles, TRUE, INFINITE);
printf("\n");
printf("Starting Thread Test\n");
printf("at %02d/%02d/%02d %02d:%02d:%02d\n", local_time.wYear, local_time.wMonth, local_time.wDay, local_time.wHour, local_time.wMinute, local_time.wSecond);
printf("\n");
CloseHandle(thread1Handle);
CloseHandle(thread2Handle);
LARGE_INTEGER Frequency;
LARGE_INTEGER StartingTime, EndingTime, ElapsedMicroseconds;
QueryPerformanceFrequency(&Frequency);
QueryPerformanceCounter(&StartingTime);
return 0;
uint32_t success_counter = 0;
for (int i = 0; i < loops; i++)
{
Run_Threads(&increment_pipe1, &increment_pipe2, &increment_pipe3, &increment_pipe4, &increment_pipe5);
//if (ringbuffer_valid_and_equal(check_pipe1.input_connector->connection[0], check_pipe2.input_connector->connection[0], hFile))
if (ringbuffer_valid_and_equal(check_pipe1.input_connector->connection[0], check_pipe2.input_connector->connection[0]))
success_counter++;
}
QueryPerformanceCounter(&EndingTime);
ElapsedMicroseconds.QuadPart = EndingTime.QuadPart - StartingTime.QuadPart;
ElapsedMicroseconds.QuadPart *= 1000000;
ElapsedMicroseconds.QuadPart /= Frequency.QuadPart;
int seconds = ElapsedMicroseconds.QuadPart / 1000000;
int hours = seconds / 3600;
int minutes = (seconds - (hours * 3600)) / 60;
seconds = seconds - (minutes * 60);
printf("Loops: %d\n", loops);
printf("Successfull: %d\n", success_counter);
printf("Failed: %d\n", loops - success_counter);
printf("Count of Concurrent Data Operations: %d\n", 2 * NUMBER_OF_ELEMENTS * loops);
printf("\n");
GetLocalTime(&local_time);
printf("Test finished\n");
printf("at %02d/%02d/%02d %02d:%02d:%02d\n", local_time.wYear, local_time.wMonth, local_time.wDay, local_time.wHour, local_time.wMinute, local_time.wSecond);
printf("Elapsed time: %02d:%02d:%02d", hours, minutes, seconds);
printf("\n");
//CloseHandle(hFile);
}