Skip to content

Commit e2efee6

Browse files
whollowhollo
authored
Merge branch 'FlagOpen:main' into Fix-llama3-train-debug-DP2
2 parents a095423 + 3ea0baf commit e2efee6

9 files changed

Lines changed: 32 additions & 35 deletions

File tree

‎.github/workflows/test.yml‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -53,12 +53,9 @@ jobs:
5353
if [ -n "$GITHUB_HEAD_REF" ] && [ -n "$GITHUB_BASE_REF" ]; then
5454
from_ref="origin/$GITHUB_HEAD_REF"
5555
to_ref="origin/$GITHUB_BASE_REF"
56-
else
57-
from_ref="HEAD^"
58-
to_ref="HEAD"
56+
echo "From reference: $from_ref; To reference: $to_ref"
57+
pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
5958
fi
60-
echo "From reference: $from_ref; To reference: $to_ref"
61-
pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
6259
continue-on-error: false
6360

6461
- name: Check the current working directory

‎.github/workflows/torch-api-test.yml‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -50,12 +50,9 @@ jobs:
5050
if [ -n "$GITHUB_HEAD_REF" ] && [ -n "$GITHUB_BASE_REF" ]; then
5151
from_ref="origin/$GITHUB_HEAD_REF"
5252
to_ref="origin/$GITHUB_BASE_REF"
53-
else
54-
from_ref="HEAD^"
55-
to_ref="HEAD"
53+
echo "From reference: $from_ref; To reference: $to_ref"
54+
pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
5655
fi
57-
echo "From reference: $from_ref; To reference: $to_ref"
58-
pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
5956
continue-on-error: false
6057

6158
- name: Run `make` to build the project

‎.github/workflows/unit-test.yml‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -53,12 +53,9 @@ jobs:
5353
if [ -n "$GITHUB_HEAD_REF" ] && [ -n "$GITHUB_BASE_REF" ]; then
5454
from_ref="origin/$GITHUB_HEAD_REF"
5555
to_ref="origin/$GITHUB_BASE_REF"
56-
else
57-
from_ref="HEAD^"
58-
to_ref="HEAD"
56+
echo "From reference: $from_ref; To reference: $to_ref"
57+
pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
5958
fi
60-
echo "From reference: $from_ref; To reference: $to_ref"
61-
pre-commit run --from-ref "$from_ref" --to-ref "$to_ref"
6259
continue-on-error: false
6360

6461
- name: Build Google Test

‎flagcx/core/group.cc‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -109,10 +109,8 @@ static flagcxResult_t groupLaunch(struct flagcxAsyncJob *job_) {
109109

110110
// Each groupLaunch we create a semaphore to track the p2p ops
111111
// and a stream to launch host or device func
112-
struct flagcxHostSemaphore *semaphore;
113-
FLAGCXCHECK(flagcxCalloc(&semaphore, 1));
114-
semaphore->flag = 0;
115-
semaphore->counter = 0;
112+
std::shared_ptr<flagcxHostSemaphore> semaphore =
113+
std::make_shared<flagcxHostSemaphore>();
116114
flagcxStream_t launchStream = nullptr;
117115

118116
if (groupCommPreconnectHeadMain != nullptr) {
@@ -312,7 +310,7 @@ static flagcxResult_t groupLaunch(struct flagcxAsyncJob *job_) {
312310
}
313311
} else {
314312
FLAGCXCHECK(deviceAdaptor->launchHostFunc(launchStream, cpuAsyncKernel,
315-
(void *)semaphore));
313+
(void *)semaphore.get()));
316314
}
317315
// deprecated code path for host func, since the previous
318316
// hang issue may be walked around by using zero copy

‎flagcx/core/launch_kernel.cc‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -50,8 +50,7 @@ void cpuAsyncLoadWithMaxSpinCount(void *args) {
5050

5151
void cpuAsyncKernel(void *args) {
5252
flagcxHostSemaphore *semaphore = (flagcxHostSemaphore *)args;
53-
semaphore->signalFlag();
53+
semaphore->signalStart();
5454
semaphore->wait();
55-
free(semaphore);
56-
semaphore = nullptr;
55+
semaphore->signalEnd();
5756
}

‎flagcx/core/launch_kernel.h‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -36,28 +36,33 @@ void cpuAsyncLoad(void *args);
3636
void cpuAsyncLoadWithMaxSpinCount(void *args);
3737

3838
struct flagcxHostSemaphore {
39-
int flag; // if ready to be triggered
40-
int counter; // total operations to wait for inside the group
39+
int start = 0; // started or not
40+
int end = 0; // ended or not
41+
int counter = 0; // total operations to wait for inside the group
4142
std::vector<flagcxEvent_t> events;
4243

44+
~flagcxHostSemaphore() {
45+
for (auto event : events) {
46+
deviceAdaptor->eventDestroy(event);
47+
}
48+
}
4349
flagcxEvent_t getEvent() {
4450
events.push_back(nullptr);
4551
auto &event = events.back();
4652
deviceAdaptor->eventCreate(&event);
4753
return event;
4854
}
49-
void signalFlag() { __atomic_store_n(&flag, 1, __ATOMIC_RELEASE); }
55+
void signalStart() { __atomic_store_n(&start, 1, __ATOMIC_RELEASE); }
56+
void signalEnd() { __atomic_store_n(&end, 1, __ATOMIC_RELEASE); }
5057
void signalCounter(int value) {
5158
__atomic_fetch_sub(&counter, value, __ATOMIC_RELEASE);
5259
}
53-
int poll() { return __atomic_load_n(&flag, __ATOMIC_ACQUIRE); }
60+
int pollStart() { return __atomic_load_n(&start, __ATOMIC_ACQUIRE); }
61+
int pollEnd() { return __atomic_load_n(&end, __ATOMIC_ACQUIRE); }
5462
void wait() {
5563
while (__atomic_load_n(&counter, __ATOMIC_ACQUIRE) > 0) {
5664
sched_yield();
5765
}
58-
for (auto event : events) {
59-
deviceAdaptor->eventDestroy(event);
60-
}
6166
}
6267
};
6368

‎flagcx/core/net.cc‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -136,7 +136,7 @@ flagcxResult_t flagcxNetInit(struct flagcxHeteroComm *comm) {
136136

137137
flagcxResult_t flagcxProxySend(sendNetResources *resources, void *data,
138138
size_t size, flagcxProxyArgs *args) {
139-
if (!args->semaphore->poll()) {
139+
if (!args->semaphore->pollStart()) {
140140
return flagcxSuccess;
141141
}
142142
if (args->transmitted < args->chunkSteps) {
@@ -220,7 +220,7 @@ flagcxResult_t flagcxProxySend(sendNetResources *resources, void *data,
220220

221221
flagcxResult_t flagcxProxyRecv(recvNetResources *resources, void *data,
222222
size_t size, flagcxProxyArgs *args) {
223-
if (!args->semaphore->poll()) {
223+
if (!args->semaphore->pollStart()) {
224224
return flagcxSuccess;
225225
}
226226
if (args->copied < args->chunkSteps) {

‎flagcx/core/proxy.cc‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -196,7 +196,8 @@ static flagcxResult_t progressOps(struct flagcxProxyState *proxyState,
196196
}
197197
}
198198
} else {
199-
if (op->args.done == 1) {
199+
if (op->args.done == 1 && op->args.semaphore->pollEnd()) {
200+
op->args.semaphore.reset();
200201
flagcxIntruQueueDelete(queue, op);
201202
free(op);
202203
}
@@ -220,7 +221,9 @@ static flagcxResult_t progressOps(struct flagcxProxyState *proxyState,
220221
}
221222
}
222223
} else {
223-
if (op->args.done == 1) {
224+
if (op->args.done == 1 && op->args.semaphore->pollEnd()) {
225+
// update refcount and delete semaphore when refcount = 0
226+
op->args.semaphore.reset();
224227
flagcxIntruQueueDelete(queue, op);
225228
free(op);
226229
}

‎flagcx/core/proxy.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
#include "net.h"
1616
#include "reg_pool.h"
1717
#include "socket.h"
18+
#include <memory>
1819
#include <pthread.h>
1920

2021
enum flagcxProxyOpState {
@@ -117,7 +118,7 @@ struct flagcxProxyArgs {
117118

118119
/*for launch*/
119120
int deviceFuncRelaxedOrdering = 0;
120-
struct flagcxHostSemaphore *semaphore = nullptr;
121+
std::shared_ptr<flagcxHostSemaphore> semaphore;
121122
// only for device func, to be deprecated
122123
volatile bool eventRecorded = false;
123124
volatile bool hlArgs = false;

0 commit comments

Comments
 (0)